The FASTEN server can now override the default topic of Kafka producers.
@@ -28,9 +28,7 @@ import eu.fasten.core.plugins.KafkaConsumer;
import eu.fasten.core.plugins.KafkaProducer;
import eu.fasten.server.kafka.FastenKafkaConsumer;
import eu.fasten.server.kafka.FastenKafkaProducer;
-import org.jooq.tools.json.JSONParser;
-import org.jooq.tools.json.ParseException;
-import org.jooq.tools.json.JSONObject;
+import org.json.JSONObject;
import org.pf4j.JarPluginManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,7 +37,9 @@ import picocli.CommandLine.Option;
import java.io.FileReader;
import java.io.IOException;
+import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.Paths;
import java.sql.SQLException;
import java.util.List;
import java.util.stream.Collectors;
@@ -126,17 +126,22 @@ public class FastenServer implements Runnable {
// Change the default topics of the plug-ins if a JSON file of the topics is given
if(pluginTopic != null) {
- JSONParser parser = new JSONParser();
try {
- JSONObject jsonObject = (JSONObject) parser.parse(new FileReader(pluginTopic));
+ JSONObject jsonObject = new JSONObject(new String(Files.readAllBytes(Paths.get(pluginTopic))));
kafkaConsumers.forEach((k) -> {
- if(jsonObject.containsKey(k.getClass().getSimpleName())) {
- k.setTopic(jsonObject.get(k.getClass().getSimpleName()).toString());
+ if(jsonObject.has(k.getClass().getSimpleName())) {
+ k.setTopic(jsonObject.getJSONObject(k.getClass().getSimpleName()).get("consumer").toString());
}
});
- } catch (IOException | ParseException e) {
+ kafkaProducers.forEach((k) ->{
+ if(jsonObject.has(k.getClass().getSimpleName())) {
+ k.setProducerTopic(jsonObject.getJSONObject(k.getClass().getSimpleName()).get("producer").toString());
+ }
+ });
+
+ } catch (IOException e) {
logger.error("Failed to read the JSON file of the topics: {}", e.getMessage());
}
}
