From d12fa554286983d49a441087e5a84599e9bff29b Mon Sep 17 00:00:00 2001 From: Koji Kawamura Date: Tue, 28 Feb 2017 12:57:04 +0900 Subject: [PATCH] NIFI-3189: ConsumeKafka 0.9 and 0.10 with downstream backpressure Currently, NiFi Kafka consumer processors have following issue. While downstream connections are full, ConsumeKafka is not scheduled to run onTrigger. It stopps executing poll to tell Kafka server that this client is alive. Thus, after a while in that situation, Kafka server rebalances the client. When downstream connections back to normal, although ConsumeKafka is scheduled again, the client is no longer a part of a consumer group. If this happens, Kafka client succeeds polling messages when ConsumeKafka processor resumes, but fails to commit offset. Received messages are already committed into NiFi flow, but since consumer offset is not updated, those will be consumed again, duplicated. In order to address above issue: - For ConsumeKafka_0_10, use latest client library Above issue has been addressed by KIP-62. The latest Kafka consumer poll checks if the client instance is still valid, and rejoin the group if not, before consuming messages. - For ConsumeKafka (0.9), added manual retention logic using pause/resume Kafka client 0.9 doesn't have background thread heartbeat, so similar machanism is added manually. Use Kafka pause/resume consumer API to tell Kafka server that the client stops consuming messages but is still alive. Another internal thread is used to perform paused poll periodically based on the time passed since the last onTrigger(poll) is executed. --- .../kafka/pubsub/ConsumerLease.java | 16 ++--- .../processors/kafka/pubsub/ConsumeKafka.java | 72 +++++++++++++++++++ .../kafka/pubsub/ConsumerLease.java | 35 +++++++++ .../processors/kafka/pubsub/ConsumerPool.java | 6 ++ .../kafka/pubsub/ConsumeKafkaTest.java | 36 ++++++++++ nifi-nar-bundles/nifi-kafka-bundle/pom.xml | 2 +- 6 files changed, 157 insertions(+), 10 deletions(-) diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java index 97ebfc6ee7d1..7ea180d25333 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java @@ -130,15 +130,13 @@ public void onPartitionsAssigned(final Collection partitions) { */ void poll() { /** - * Implementation note: If we take too long (30 secs?) between kafka - * poll calls and our own record processing to any subsequent poll calls - * or the commit we can run into a situation where the commit will - * succeed to the session but fail on committing offsets. This is - * apparently different than the Kafka scenario of electing to rebalance - * for other reasons but in this case is due a session timeout. It - * appears Kafka KIP-62 aims to offer more control over the meaning of - * various timeouts. If we do run into this case it could result in - * duplicates. + * Implementation note: + * Even if ConsumeKafka is not scheduled to poll due to downstream connection back-pressure is engaged, + * for longer than session.timeout.ms (defaults to 10 sec), Kafka consumer sends heartbeat from background thread. + * If this situation lasts longer than max.poll.interval.ms (defaults to 5 min), Kafka consumer sends + * Leave Group request to Group Coordinator. When ConsumeKafka processor is scheduled again, Kafka client checks + * if this client instance is still a part of consumer group. If not, it rejoins before polling messages. + * This behavior has been fixed via Kafka KIP-62 and available from Kafka client 0.10.1.0. */ try { final ConsumerRecords records = kafkaConsumer.poll(10); diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka.java index 3e01e51ea7af..ff017461ad76 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka.java @@ -25,6 +25,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.KafkaException; @@ -36,10 +38,12 @@ import org.apache.nifi.annotation.behavior.WritesAttributes; import org.apache.nifi.annotation.documentation.CapabilityDescription; import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; import org.apache.nifi.annotation.lifecycle.OnStopped; import org.apache.nifi.annotation.lifecycle.OnUnscheduled; import org.apache.nifi.components.AllowableValue; import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; import org.apache.nifi.components.ValidationContext; import org.apache.nifi.components.ValidationResult; import org.apache.nifi.logging.ComponentLog; @@ -162,6 +166,10 @@ public class ConsumeKafka extends AbstractProcessor { private volatile ConsumerPool consumerPool = null; private final Set activeLeases = Collections.synchronizedSet(new HashSet<>()); + private int heartbeatCheckIntervalMillis; + private volatile long lastTriggeredTimestamp = -1L; + private volatile ScheduledExecutorService connectionRetainer; + static { List descriptors = new ArrayList<>(); descriptors.addAll(KafkaProcessorUtils.getCommonPropertyDescriptors()); @@ -275,8 +283,72 @@ public void interruptActiveThreads() { activeLeases.clear(); } + @OnScheduled + public void onScheduled(final ProcessContext context) { + + final PropertyValue heartbeatIntervalMsConfig = context.getProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG); + if (heartbeatIntervalMsConfig != null && heartbeatIntervalMsConfig.isSet()) { + heartbeatCheckIntervalMillis = heartbeatIntervalMsConfig.asInteger(); + } else { + // Derived from org.apache.kafka.clients.consumer.ConsumerConfig. + heartbeatCheckIntervalMillis = 3_000; + } + + // Without this, it remains -1 if downstream connections are full when this processor is scheduled at the 1st run after restart. + lastTriggeredTimestamp = System.currentTimeMillis(); + + // Stop previous connectionRetainer, if any, just in case, this shouldn't happen though + final ComponentLog logger = getLogger(); + if (connectionRetainer != null) { + logger.warn("Connection retainer {} is still running, indicating something had happened.", + new Object[]{connectionRetainer}); + stopConnectionRetainer(); + } + connectionRetainer = Executors.newSingleThreadScheduledExecutor(); + connectionRetainer.scheduleAtFixedRate(() -> { + final long now = System.currentTimeMillis(); + if (lastTriggeredTimestamp < 0 + || lastTriggeredTimestamp > now - heartbeatCheckIntervalMillis) { + if (logger.isDebugEnabled()) { + logger.debug("No need to retain connection. Triggered at {}, {} millis ago.", + new Object[]{lastTriggeredTimestamp, now - lastTriggeredTimestamp}); + } + return; + } + try { + final ConsumerPool pool = getConsumerPool(context); + if (logger.isDebugEnabled()) { + final ConsumerPool.PoolStats stats = pool.getPoolStats(); + logger.debug("Trying to retain connection. Obtained pool={}," + + " leaseObtainedCount={}, consumerCreatedCount={}, consumerClosedCount={}", + new Object[]{pool, stats.leasesObtainedCount, stats.consumerCreatedCount, stats.consumerClosedCount}); + } + pool.retainConsumers(); + } catch (final Exception e) { + logger.warn("Failed to retain connection due to {}", new Object[]{e}, e); + } + }, heartbeatCheckIntervalMillis, heartbeatCheckIntervalMillis, TimeUnit.MILLISECONDS); + } + + @OnUnscheduled + public void stopConnectionRetainer() { + if (connectionRetainer != null) { + final ComponentLog logger = getLogger(); + logger.debug("Canceling connectionRetainer... {}", new Object[]{connectionRetainer}); + try { + connectionRetainer.shutdownNow(); + } catch (final Exception e) { + logger.warn("Failed to shutdown connection retainer {} due to {}", new Object[]{connectionRetainer, e}, e); + } + connectionRetainer = null; + } + } + @Override public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + + lastTriggeredTimestamp = System.currentTimeMillis(); + final ConsumerPool pool = getConsumerPool(context); if (pool == null) { context.yield(); diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java index cd9365dbe739..53e7a23752a9 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java @@ -23,7 +23,9 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; import javax.xml.bind.DatatypeConverter; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; @@ -64,6 +66,7 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe private long leaseStartNanos = -1; private boolean lastPollEmpty = false; private int totalFlowFiles = 0; + private ReentrantLock pollingLock = new ReentrantLock(); ConsumerLease( final long maxWaitMillis, @@ -139,7 +142,9 @@ void poll() { * appears Kafka KIP-62 aims to offer more control over the meaning of * various timeouts. If we do run into this case it could result in * duplicates. + * This can be avoided by calling retainConnection() periodically. */ + pollingLock.lock(); try { final ConsumerRecords records = kafkaConsumer.poll(10); lastPollEmpty = records.count() == 0; @@ -147,6 +152,36 @@ void poll() { } catch (final Throwable t) { this.poison(); throw t; + } finally { + pollingLock.unlock(); + } + } + + /** + * Execute poll using pause API just for sending heartbeat, not polling messages. + */ + void retainConnection() { + pollingLock.lock(); + TopicPartition[] assignments = null; + try { + final Set assignmentSet = kafkaConsumer.assignment(); + if (assignmentSet.isEmpty()) { + return; + } + if (logger.isDebugEnabled()) { + logger.debug("Pausing " + assignmentSet); + } + assignments = assignmentSet.toArray(new TopicPartition[assignmentSet.size()]); + kafkaConsumer.pause(assignments); + kafkaConsumer.poll(0); + if (logger.isDebugEnabled()) { + logger.debug("Resuming " + assignments); + } + } finally { + if (assignments != null) { + kafkaConsumer.resume(assignments); + } + pollingLock.unlock(); } } diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerPool.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerPool.java index e13a8c3c4d1a..faa4643e9ee5 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerPool.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerPool.java @@ -22,6 +22,7 @@ import java.io.Closeable; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; @@ -127,6 +128,11 @@ public ConsumerLease obtainConsumer(final ProcessSession session) { return lease; } + public void retainConsumers() { + Arrays.stream(pooledLeases.toArray(new ConsumerLease[]{})) + .forEach(ConsumerLease::retainConnection); + } + /** * Exposed as protected method for easier unit testing * diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java index f6febd2ac351..ce1092398eb9 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java @@ -20,8 +20,10 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.ByteArrayDeserializer; +import org.apache.nifi.components.PropertyValue; import org.apache.nifi.logging.ComponentLog; import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; import org.junit.Test; @@ -29,6 +31,7 @@ import static org.junit.Assert.fail; import org.junit.Before; import static org.mockito.Matchers.anyObject; +import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -164,4 +167,37 @@ protected ConsumerPool createConsumerPool(final ProcessContext context, final Co verifyNoMoreInteractions(mockLease); } + @Test + public void validateConsumerRetainer() throws Exception { + final ConsumerPool consumerPool = mock(ConsumerPool.class); + + final ConsumeKafka processor = new ConsumeKafka() { + @Override + protected ConsumerPool createConsumerPool(ProcessContext context, ComponentLog log) { + return consumerPool; + } + }; + + final ComponentLog logger = mock(ComponentLog.class); + final ProcessorInitializationContext initializationContext = mock(ProcessorInitializationContext.class); + when(initializationContext.getLogger()).thenReturn(logger); + processor.initialize(initializationContext); + + final ProcessContext processContext = mock(ProcessContext.class); + final PropertyValue heartbeatInternalMsConfig = mock(PropertyValue.class); + when(heartbeatInternalMsConfig.isSet()).thenReturn(true); + when(heartbeatInternalMsConfig.asInteger()).thenReturn(100); + when(processContext.getProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG)).thenReturn(heartbeatInternalMsConfig); + processor.onScheduled(processContext); + + // retainConsumers should be called at least 1 time if it passed longer than heartbeat interval milliseconds. + Thread.sleep(200); + verify(consumerPool, atLeast(1)).retainConsumers(); + + processor.stopConnectionRetainer(); + + // After stopping connection retainer, it shouldn't interact with consumerPool. + Thread.sleep(200); + verifyNoMoreInteractions(consumerPool); + } } diff --git a/nifi-nar-bundles/nifi-kafka-bundle/pom.xml b/nifi-nar-bundles/nifi-kafka-bundle/pom.xml index 99d804639cb0..130609d2e65f 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/pom.xml +++ b/nifi-nar-bundles/nifi-kafka-bundle/pom.xml @@ -25,7 +25,7 @@ 0.8.2.2 0.9.0.1 - 0.10.0.1 + 0.10.2.0