Set comsumerTopic and pluginError in standard way.
@@ -49,6 +49,7 @@ public class VulnerabilityPackagesListener extends Plugin {
public static class VulnerabilityPackagesKafkaPlugin implements KafkaPlugin, DBConnector {
private static Map<String, DSLContext> contexts;
+ private String consumeTopic = "fasten.CallableIndexFastenPlugin.out";
private Exception pluginError = null;
private final Gson gson = new Gson();
private final static Logger logger = LoggerFactory.getLogger("VulnerabilityPackagesKafkaPlugin");
@@ -83,6 +84,11 @@ public class VulnerabilityPackagesListener extends Plugin {
public void stop() {
}
+ @Override
+ public void setPluginError(Exception throwable) {
+ this.pluginError = throwable;
+ }
+
@Override
public Exception getPluginError() {
return pluginError;
@@ -94,13 +100,12 @@ public class VulnerabilityPackagesListener extends Plugin {
@Override
public Optional<List<String>> consumeTopic() {
- String consumerTopic = "fasten.CallableIndexFastenPlugin.out";
- return Optional.of(Collections.singletonList(consumerTopic));
+ return Optional.of(Collections.singletonList(consumeTopic));
}
@Override
public void setTopic(String topicName) {
-
+ consumeTopic = topicName;
}
@Override
@@ -128,7 +133,7 @@ public class VulnerabilityPackagesListener extends Plugin {
catch (Exception e) {
var error = "Error processing package update: " + e;
logger.error(error);
- setPluginError(new PluginRuntimeException(error));
+ setPluginError(e);
}
}
