You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2017/03/30 20:30:08 UTC
nifi git commit: NIFI-3189: ConsumeKafka 0.9 and 0.10 with downstream
backpressure
Repository: nifi
Updated Branches:
refs/heads/master 282e1a7b1 -> fd92999da
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.
This closes #1527.
Signed-off-by: Bryan Bende <bb...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/fd92999d
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/fd92999d
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/fd92999d
Branch: refs/heads/master
Commit: fd92999dafc040940011c87bb2ee2c8edf5f96a2
Parents: 282e1a7
Author: Koji Kawamura <ij...@apache.org>
Authored: Tue Feb 28 12:57:04 2017 +0900
Committer: Bryan Bende <bb...@apache.org>
Committed: Thu Mar 30 16:29:40 2017 -0400
----------------------------------------------------------------------
.../processors/kafka/pubsub/ConsumerLease.java | 16 ++---
.../processors/kafka/pubsub/ConsumeKafka.java | 72 ++++++++++++++++++++
.../processors/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(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/fd92999d/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java
----------------------------------------------------------------------
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 97ebfc6..7ea180d 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 abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe
*/
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<byte[], byte[]> records = kafkaConsumer.poll(10);
http://git-wip-us.apache.org/repos/asf/nifi/blob/fd92999d/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka.java
----------------------------------------------------------------------
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 3e01e51..ff01746 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.HashSet;
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.WritesAttribute;
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<ConsumerLease> activeLeases = Collections.synchronizedSet(new HashSet<>());
+ private int heartbeatCheckIntervalMillis;
+ private volatile long lastTriggeredTimestamp = -1L;
+ private volatile ScheduledExecutorService connectionRetainer;
+
static {
List<PropertyDescriptor> descriptors = new ArrayList<>();
descriptors.addAll(KafkaProcessorUtils.getCommonPropertyDescriptors());
@@ -275,8 +283,72 @@ public class ConsumeKafka extends AbstractProcessor {
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();
http://git-wip-us.apache.org/repos/asf/nifi/blob/fd92999d/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java
----------------------------------------------------------------------
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 cd9365d..53e7a23 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.Collection;
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 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe
* 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<byte[], byte[]> records = kafkaConsumer.poll(10);
lastPollEmpty = records.count() == 0;
@@ -147,6 +152,36 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe
} 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<TopicPartition> 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();
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/fd92999d/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerPool.java
----------------------------------------------------------------------
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 e13a8c3..faa4643 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 org.apache.nifi.logging.ComponentLog;
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 class ConsumerPool implements Closeable {
return lease;
}
+ public void retainConsumers() {
+ Arrays.stream(pooledLeases.toArray(new ConsumerLease[]{}))
+ .forEach(ConsumerLease::retainConnection);
+ }
+
/**
* Exposed as protected method for easier unit testing
*
http://git-wip-us.apache.org/repos/asf/nifi/blob/fd92999d/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
----------------------------------------------------------------------
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 f6febd2..ce10923 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.Consumer;
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.assertTrue;
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 @@ public class ConsumeKafkaTest {
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);
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/fd92999d/nifi-nar-bundles/nifi-kafka-bundle/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/pom.xml b/nifi-nar-bundles/nifi-kafka-bundle/pom.xml
index 99d8046..130609d 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-kafka-bundle/pom.xml
@@ -25,7 +25,7 @@
<properties>
<kafka8.version>0.8.2.2</kafka8.version>
<kafka9.version>0.9.0.1</kafka9.version>
- <kafka10.version>0.10.0.1</kafka10.version>
+ <kafka10.version>0.10.2.0</kafka10.version>
</properties>
<modules>