Process the whole working set rather than one of its items in the FASTEN server
@@ -266,7 +266,8 @@ public class FastenKafkaPlugin implements FastenServerPlugin {
}
private void handleWorkingSet(KafkaRecordKind kafkaRecordKind) {
- if (plugin.getWorkingSet().isPresent() && !plugin.getWorkingSet().get().isEmpty()) {
+ if (plugin.getWorkingSet().isPresent()) {
+ while (!plugin.getWorkingSet().get().isEmpty()) {
// The plug-in might spend a lot of time in processing its working set,
// therefore, we need to keep the Kafka consumers alive
sendHeartBeat(connNorm);
@@ -279,6 +280,7 @@ public class FastenKafkaPlugin implements FastenServerPlugin {
logger.info("Successfully processed working set message from " + kafkaRecordKind.toString() + " topic");
}
}
+ }
/**
* Consumer strategy (using local storage):
