Adds a Kafka connection to the FASTEN server for putting erorrs into a Kafka topic.
@@ -2,9 +2,13 @@ package eu.fasten.server.kafka;
import eu.fasten.core.plugins.KafkaConsumer;
import org.apache.kafka.clients.consumer.CommitFailedException;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -17,6 +21,9 @@ import java.util.concurrent.CountDownLatch;
public class FastenKafkaConsumer extends FastenKafkaConnection {
private final Logger logger = LoggerFactory.getLogger(FastenKafkaConsumer.class.getName());
+ // This produces errors of a plug-in into a Kafka topic.
+ private KafkaProducer errorLog;
+
private org.apache.kafka.clients.consumer.KafkaConsumer<String, String> connection;
private KafkaConsumer<String> kafkaConsumer;
private CountDownLatch mLatch;
@@ -24,6 +31,8 @@ public class FastenKafkaConsumer extends FastenKafkaConnection {
public FastenKafkaConsumer(Properties p, KafkaConsumer kc) {
super(p);
this.kafkaConsumer = kc;
+ this.setErrorLogConn(p.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), kc.getClass().getCanonicalName());
+
this.mLatch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
@@ -41,6 +50,20 @@ public class FastenKafkaConsumer extends FastenKafkaConnection {
}
+ /*
+ This methods sets up a connection for producing error logs of a plug-in into a Kafka topic.
+ */
+ private void setErrorLogConn(String serverAddress, String clientID){
+
+ Properties p = new Properties();
+ p.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, serverAddress);
+ p.setProperty(ProducerConfig.CLIENT_ID_CONFIG, clientID);
+ p.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ p.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+
+ this.errorLog = new KafkaProducer(p);
+ }
+
@Override
public void run() {
logger.debug("Starting consumer: {}", kafkaConsumer.getClass());
@@ -54,10 +77,10 @@ public class FastenKafkaConsumer extends FastenKafkaConnection {
ConsumerRecords<String, String> records = connection.poll(Duration.ofMillis(100));
List<String> topics = kafkaConsumer.consumerTopics();
- for(ConsumerRecord<String, String> r : records) System.out.println(r.key() + " " + r.value());
+ //for(ConsumerRecord<String, String> r : records) System.out.println(r.key() + " " + r.value());
for (String topic : topics){
- for(ConsumerRecord<String, String> r : records.records(topic)) System.out.println("K: " + r.key());
+ //for(ConsumerRecord<String, String> r : records.records(topic)) System.out.println("K: " + r.key());
records.records(topic).forEach(r -> kafkaConsumer.consume(topic, r));
doCommitSync();
}
@@ -78,6 +101,7 @@ public class FastenKafkaConsumer extends FastenKafkaConnection {
private void doCommitSync() {
try {
connection.commitSync();
+ logger.debug("Committed the processed record...");
} catch (WakeupException e) {
// we're shutting down, but finish the commit first and then
// rethrow the exception so that the main loop can exit
