You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rh...@apache.org on 2019/10/15 21:27:11 UTC
[kafka] branch 2.2 updated: KAFKA-9014: Fix AssertionError when
SourceTask.poll returns an empty list (#7491)
This is an automated email from the ASF dual-hosted git repository.
rhauch pushed a commit to branch 2.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.2 by this push:
new f9c8a58 KAFKA-9014: Fix AssertionError when SourceTask.poll returns an empty list (#7491)
f9c8a58 is described below
commit f9c8a580ba7482181d76af267e6f7344811de871
Author: Konstantine Karantasis <ko...@confluent.io>
AuthorDate: Tue Oct 15 14:08:31 2019 -0700
KAFKA-9014: Fix AssertionError when SourceTask.poll returns an empty list (#7491)
Author: Konstantine Karantasis <ko...@confluent.io>
Reviewer: Randall Hauch <rh...@gmail.com>
---
.../kafka/connect/runtime/WorkerSourceTask.java | 3 +-
.../connect/runtime/WorkerSourceTaskTest.java | 64 ++++++++++++++++++++++
2 files changed, 66 insertions(+), 1 deletion(-)
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
index 097ac3d..acdb287 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
@@ -300,7 +300,8 @@ class WorkerSourceTask extends WorkerTask {
private boolean sendRecords() {
int processed = 0;
recordBatch(toSend.size());
- final SourceRecordWriteCounter counter = new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup);
+ final SourceRecordWriteCounter counter =
+ toSend.size() > 0 ? new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup) : null;
for (final SourceRecord preTransformRecord : toSend) {
maybeThrowProducerSendException();
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
index 1f5cc43..137cbc2 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
@@ -77,6 +77,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@PowerMockIgnore({"javax.management.*",
+ "org.apache.log4j.*",
"org.apache.kafka.connect.runtime.isolation.*"})
@RunWith(PowerMockRunner.class)
public class WorkerSourceTaskTest extends ThreadedTest {
@@ -340,6 +341,51 @@ public class WorkerSourceTaskTest extends ThreadedTest {
}
@Test
+ public void testPollReturnsNoRecords() throws Exception {
+ // Test that the task handles an empty list of records
+ createWorkerTask();
+
+ sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
+ EasyMock.expectLastCall();
+ sourceTask.start(TASK_PROPS);
+ EasyMock.expectLastCall();
+ statusListener.onStartup(taskId);
+ EasyMock.expectLastCall();
+
+ // We'll wait for some data, then trigger a flush
+ final CountDownLatch pollLatch = expectEmptyPolls(1, new AtomicInteger());
+ expectOffsetFlush(true);
+
+ sourceTask.stop();
+ EasyMock.expectLastCall();
+ expectOffsetFlush(true);
+
+ statusListener.onShutdown(taskId);
+ EasyMock.expectLastCall();
+
+ producer.close(EasyMock.anyObject(Duration.class));
+ EasyMock.expectLastCall();
+
+ transformationChain.close();
+ EasyMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ workerTask.initialize(TASK_CONFIG);
+ Future<?> taskFuture = executor.submit(workerTask);
+
+ assertTrue(awaitLatch(pollLatch));
+ assertTrue(workerTask.commitOffsets());
+ workerTask.stop();
+ assertTrue(workerTask.awaitStop(1000));
+
+ taskFuture.get();
+ assertPollMetrics(0);
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
public void testCommit() throws Exception {
// Test that the task commits properly when prompted
createWorkerTask();
@@ -682,6 +728,24 @@ public class WorkerSourceTaskTest extends ThreadedTest {
assertEquals(1800.0, metrics.currentMetricValueAsDouble(group1.metricGroup(), "source-record-active-count"), 0.001d);
}
+ private CountDownLatch expectEmptyPolls(int minimum, final AtomicInteger count) throws InterruptedException {
+ final CountDownLatch latch = new CountDownLatch(minimum);
+ // Note that we stub these to allow any number of calls because the thread will continue to
+ // run. The count passed in + latch returned just makes sure we get *at least* that number of
+ // calls
+ EasyMock.expect(sourceTask.poll())
+ .andStubAnswer(new IAnswer<List<SourceRecord>>() {
+ @Override
+ public List<SourceRecord> answer() throws Throwable {
+ count.incrementAndGet();
+ latch.countDown();
+ Thread.sleep(10);
+ return Collections.emptyList();
+ }
+ });
+ return latch;
+ }
+
private CountDownLatch expectPolls(int minimum, final AtomicInteger count) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(minimum);
// Note that we stub these to allow any number of calls because the thread will continue to