The FASTEN plug-ins have now a separate Kafka topic for their errors.
@@ -303,8 +303,8 @@ public class FastenKafkaConsumer extends FastenKafkaConnection {
logToKafka(this.serverLog, this.serverLogTopic, new Date() + "| [" + this.consumerHostName + "] Plug-in " + kafkaConsumer.getClass().getSimpleName() +
" processed successfully record [in " + timeFormatter.format((System.currentTimeMillis() - startTime) / 1000d) + " sec.]: " + r.key());
} else {
- logToKafka(this.failedRecords, this.failedRecordsTopic, generateRecordStatus(kafkaConsumer.getClass().getSimpleName(),
- r, this.kafkaConsumer.getPluginError()));
+ logToKafka(this.failedRecords, String.format("fasten.%s.STDERR", kafkaConsumer.getClass().getSimpleName()),
+ generateRecordStatus(kafkaConsumer.getClass().getSimpleName(), r, this.kafkaConsumer.getPluginError()));
}
//kafkaConsumer.freeResource();
}
