From 0956c0ad5add402d69ed7a7a23c24c1dacf63ace Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Mon, 8 Jun 2015 10:10:58 -0400 Subject: [PATCH] INTEXT-172 Make potentially long-running tasks SchedulingAwareRunnable --- .../listener/KafkaMessageListenerContainer.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java index 1b309906de..1440a3f62f 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/listener/KafkaMessageListenerContainer.java @@ -45,6 +45,7 @@ import org.springframework.integration.kafka.core.KafkaMessageBatch; import org.springframework.integration.kafka.core.KafkaTemplate; import org.springframework.integration.kafka.core.Partition; import org.springframework.integration.kafka.core.Result; +import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -335,7 +336,7 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { /** * Fetches data from Kafka for a group of partitions, located on the same broker. */ - public class FetchTask implements Runnable { + public class FetchTask implements SchedulingAwareRunnable { private final BrokerAddress brokerAddress; @@ -343,6 +344,11 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { this.brokerAddress = brokerAddress; } + @Override + public boolean isLongLived() { + return true; + } + @Override public void run() { boolean wasInterrupted = false; @@ -453,13 +459,18 @@ public class KafkaMessageListenerContainer implements SmartLifecycle { } } - private class UpdateLeadersTask implements Runnable { + private class UpdateLeadersTask implements SchedulingAwareRunnable { private final Iterable partitionsToReset; public UpdateLeadersTask(Iterable partitionsToReset) { this.partitionsToReset = partitionsToReset; } + @Override + public boolean isLongLived() { + return true; + } + @Override public void run() { // fetch can complete successfully or unsuccessfully