Fix the heartbeat call crashing the server plugin: check if consumer
is subscribed to the heartbeat topic(s).
@@ -553,14 +553,17 @@ public class FastenKafkaPlugin implements FastenServerPlugin {
/**
* This method can be used to simulate Kafka's heartbeat to avoid the eviction
* of a consumer.
+ *
+ * See https://stackoverflow.com/a/43722731
*/
private static void sendHeartBeat(KafkaConsumer<String, String> kafkaConn) {
- // See https://stackoverflow.com/a/43722731
+ if(!kafkaConn.subscription().isEmpty()) {
var currentlyAssignedPartitions = kafkaConn.assignment();
kafkaConn.pause(currentlyAssignedPartitions);
kafkaConn.poll(Duration.ZERO);
kafkaConn.resume(currentlyAssignedPartitions);
}
+ }
/**
* This is a dummy poll method for calling lazy methods such as seek.
