You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "hgeraldino (via GitHub)" <gi...@apache.org> on 2024/02/05 03:34:30 UTC
[PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
hgeraldino opened a new pull request, #15316:
URL: https://github.com/apache/kafka/pull/15316
Last batch for migrating WorkerSinkTaskTest tests to Mockito
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #15316:
URL: https://github.com/apache/kafka/pull/15316#issuecomment-1981533086
Test failures appear unrelated, and the runtime tests pass locally for me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 merged PR #15316:
URL: https://github.com/apache/kafka/pull/15316
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on code in PR #15316:
URL: https://github.com/apache/kafka/pull/15316#discussion_r1488348261
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -447,6 +457,85 @@ public void testPollRedelivery() {
assertSinkMetricValue("offset-commit-completion-total", 1.0);
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPollRedeliveryWithConsumerRebalance() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ Set<TopicPartition> newAssignment = new HashSet<>(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3));
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(newAssignment, newAssignment, newAssignment)
+ .thenReturn(Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(1))
+ // Empty consumer poll (all partitions are paused) with rebalance; one new partition is assigned
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(0))
+ // Non-empty consumer poll; all initially-assigned partitions are revoked in rebalance, and new partitions are allowed to resume
+ .thenAnswer(invocation -> {
+ ConsumerRecord<byte[], byte[]> newRecord = new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET, RAW_KEY, RAW_VALUE);
+
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptyList());
+ return new ConsumerRecords<>(Collections.singletonMap(TOPIC_PARTITION3, Collections.singletonList(newRecord)));
Review Comment:
in the original test:
This confuses me. The earlier 3rd poll assigned topic-partition 3, and iteration 3 paused the newAssignment, including topic-partition 3. Yet in the 5th poll, this record is returned for a partition which should be paused.
The 5th iteration has the put succeed, and then the partitions are unpaused. But this unpause should only allow for new records to appear in the 6th poll, right?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
Review Comment:
nit: this can be doNothing, here and wherever else the doAnswer is a noop.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -642,6 +1292,228 @@ public void testTaskCancelPreventsFinalOffsetCommit() {
verify(sinkTask).close(any());
}
+ // Verify that when commitAsync is called but the supplied callback is not called by the consumer before a
+ // rebalance occurs, the async callback does not reset the last committed offset from the rebalance.
+ // See KAFKA-5731 for more information.
+ @Test
+ @SuppressWarnings("unchecked")
Review Comment:
this suppression can be removed.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
Review Comment:
nit: the labels for each iteration should go below verification for the previous iteration and before the stubbing for the next iteration
```suggestion
// iter 3
```
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -447,6 +457,85 @@ public void testPollRedelivery() {
assertSinkMetricValue("offset-commit-completion-total", 1.0);
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPollRedeliveryWithConsumerRebalance() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ Set<TopicPartition> newAssignment = new HashSet<>(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3));
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(newAssignment, newAssignment, newAssignment)
+ .thenReturn(Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(1))
+ // Empty consumer poll (all partitions are paused) with rebalance; one new partition is assigned
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(0))
+ // Non-empty consumer poll; all initially-assigned partitions are revoked in rebalance, and new partitions are allowed to resume
+ .thenAnswer(invocation -> {
+ ConsumerRecord<byte[], byte[]> newRecord = new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET, RAW_KEY, RAW_VALUE);
+
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptyList());
+ return new ConsumerRecords<>(Collections.singletonMap(TOPIC_PARTITION3, Collections.singletonList(newRecord)));
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // If a retriable exception is thrown, we should redeliver the same batch, pausing the consumer in the meantime
+ doThrow(new RetriableException("retry"))
+ .doThrow(new RetriableException("retry"))
+ .doThrow(new RetriableException("retry"))
+ .doThrow(new RetriableException("retry"))
+ .when(sinkTask).put(any(Collection.class));
+
+ workerTask.iteration();
+ verify(consumer).pause(INITIAL_ASSIGNMENT);
+
+ // Pause
+ workerTask.iteration();
+
+ workerTask.iteration();
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+ // All partitions are re-paused in order to pause any newly-assigned partitions so that redelivery efforts can continue
+ verify(consumer).pause(newAssignment);
+
+ workerTask.iteration();
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = INITIAL_ASSIGNMENT.stream()
+ .collect(Collectors.toMap(Function.identity(), tp -> new OffsetAndMetadata(FIRST_OFFSET)));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ newAssignment = Collections.singleton(TOPIC_PARTITION3);
+
+ workerTask.iteration();
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+
+ // All partitions are resumed, as all previously paused-for-redelivery partitions were revoked
+ newAssignment.forEach(tp -> {
Review Comment:
in the original test:
The 5th poll appears to revoke the initial assignment, so I would expect that only partition 3 gets resumed.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -642,6 +1292,228 @@ public void testTaskCancelPreventsFinalOffsetCommit() {
verify(sinkTask).close(any());
}
+ // Verify that when commitAsync is called but the supplied callback is not called by the consumer before a
+ // rebalance occurs, the async callback does not reset the last committed offset from the rebalance.
+ // See KAFKA-5731 for more information.
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCommitWithOutOfOrderCallback() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ // iter 1
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalance = invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ };
+
+ Answer<Set<TopicPartition>> initialAssignment = invocation -> INITIAL_ASSIGNMENT;
+ Answer<Long> partition1InitialPosition = invocation -> FIRST_OFFSET;
+ Answer<Long> partition2InitialPosition = invocation -> FIRST_OFFSET;
+
+ // iter 2
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPoll = expectConsumerPoll(1);
+
+ expectTaskGetTopic();
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final List<TopicPartition> originalPartitions = new ArrayList<>(INITIAL_ASSIGNMENT);
+ final List<TopicPartition> rebalancedPartitions = asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3);
+ final Map<TopicPartition, OffsetAndMetadata> rebalanceOffsets = new HashMap<>();
+ rebalanceOffsets.put(TOPIC_PARTITION, workerCurrentOffsets.get(TOPIC_PARTITION));
+ rebalanceOffsets.put(TOPIC_PARTITION2, workerCurrentOffsets.get(TOPIC_PARTITION2));
+ rebalanceOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> postRebalanceCurrentOffsets = new HashMap<>();
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 3));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET + 2));
+
+ Answer<Set<TopicPartition>> originalPartitionAssignment = invocation -> new HashSet<>(originalPartitions);
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that the callback is not called until after the rebalance
+ // changes the lastCommittedOffsets. To fake this for tests we have the commitAsync build a function
+ // that will call the callback with the appropriate parameters, and we'll run that function later.
+ final AtomicReference<Runnable> asyncCallbackRunner = new AtomicReference<>();
+ final AtomicBoolean asyncCallbackRan = new AtomicBoolean();
+
+ doAnswer(invocation -> {
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+ asyncCallbackRunner.set(() -> {
+ callback.onComplete(offsets, null);
+ asyncCallbackRan.set(true);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ // Expect the next poll to discover and perform the rebalance, THEN complete the previous callback handler,
+ // and then return one record for TP1 and one for TP3.
+ final AtomicBoolean rebalanced = new AtomicBoolean();
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalanced = invocation -> {
+ // Rebalance always begins with revoking current partitions ...
+ rebalanceListener.getValue().onPartitionsRevoked(originalPartitions);
+ // Respond to the rebalance
+ Map<TopicPartition, Long> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, rebalanceOffsets.get(TOPIC_PARTITION).offset());
+ offsets.put(TOPIC_PARTITION2, rebalanceOffsets.get(TOPIC_PARTITION2).offset());
+ offsets.put(TOPIC_PARTITION3, rebalanceOffsets.get(TOPIC_PARTITION3).offset());
+ sinkTaskContext.getValue().offset(offsets);
+ rebalanceListener.getValue().onPartitionsAssigned(rebalancedPartitions);
+ rebalanced.set(true);
+
+ // Run the previous async commit handler
+ asyncCallbackRunner.get().run();
+
+ // And prep the two records to return
+ long timestamp = RecordBatch.NO_TIMESTAMP;
+ TimestampType timestampType = TimestampType.NO_TIMESTAMP_TYPE;
+ List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION, FIRST_OFFSET + recordsReturnedTp1 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET + recordsReturnedTp3 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ recordsReturnedTp1 += 1;
+ recordsReturnedTp3 += 1;
+ return new ConsumerRecords<>(Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), records));
+ };
+
+ // onPartitionsRevoked
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // onPartitionsAssigned - step 1
+ final long offsetTp1 = rebalanceOffsets.get(TOPIC_PARTITION).offset();
+ final long offsetTp2 = rebalanceOffsets.get(TOPIC_PARTITION2).offset();
+ final long offsetTp3 = rebalanceOffsets.get(TOPIC_PARTITION3).offset();
+
+ // iter 4 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(postRebalanceCurrentOffsets)).thenReturn(postRebalanceCurrentOffsets);
+
+ // Setup mocks
+ when(consumer.assignment())
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenAnswer(originalPartitionAssignment)
Review Comment:
nit: These can all be INITIAL_ASSIGNMENT, you don't need these extra variables.
This makes it look like the first three are different than the next two, when they aren't actually.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -642,6 +1292,228 @@ public void testTaskCancelPreventsFinalOffsetCommit() {
verify(sinkTask).close(any());
}
+ // Verify that when commitAsync is called but the supplied callback is not called by the consumer before a
+ // rebalance occurs, the async callback does not reset the last committed offset from the rebalance.
+ // See KAFKA-5731 for more information.
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCommitWithOutOfOrderCallback() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ // iter 1
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalance = invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ };
+
+ Answer<Set<TopicPartition>> initialAssignment = invocation -> INITIAL_ASSIGNMENT;
+ Answer<Long> partition1InitialPosition = invocation -> FIRST_OFFSET;
+ Answer<Long> partition2InitialPosition = invocation -> FIRST_OFFSET;
+
+ // iter 2
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPoll = expectConsumerPoll(1);
+
+ expectTaskGetTopic();
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final List<TopicPartition> originalPartitions = new ArrayList<>(INITIAL_ASSIGNMENT);
+ final List<TopicPartition> rebalancedPartitions = asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3);
+ final Map<TopicPartition, OffsetAndMetadata> rebalanceOffsets = new HashMap<>();
+ rebalanceOffsets.put(TOPIC_PARTITION, workerCurrentOffsets.get(TOPIC_PARTITION));
+ rebalanceOffsets.put(TOPIC_PARTITION2, workerCurrentOffsets.get(TOPIC_PARTITION2));
+ rebalanceOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> postRebalanceCurrentOffsets = new HashMap<>();
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 3));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET + 2));
+
+ Answer<Set<TopicPartition>> originalPartitionAssignment = invocation -> new HashSet<>(originalPartitions);
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that the callback is not called until after the rebalance
+ // changes the lastCommittedOffsets. To fake this for tests we have the commitAsync build a function
+ // that will call the callback with the appropriate parameters, and we'll run that function later.
+ final AtomicReference<Runnable> asyncCallbackRunner = new AtomicReference<>();
+ final AtomicBoolean asyncCallbackRan = new AtomicBoolean();
+
+ doAnswer(invocation -> {
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+ asyncCallbackRunner.set(() -> {
+ callback.onComplete(offsets, null);
+ asyncCallbackRan.set(true);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ // Expect the next poll to discover and perform the rebalance, THEN complete the previous callback handler,
+ // and then return one record for TP1 and one for TP3.
+ final AtomicBoolean rebalanced = new AtomicBoolean();
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalanced = invocation -> {
+ // Rebalance always begins with revoking current partitions ...
+ rebalanceListener.getValue().onPartitionsRevoked(originalPartitions);
+ // Respond to the rebalance
+ Map<TopicPartition, Long> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, rebalanceOffsets.get(TOPIC_PARTITION).offset());
+ offsets.put(TOPIC_PARTITION2, rebalanceOffsets.get(TOPIC_PARTITION2).offset());
+ offsets.put(TOPIC_PARTITION3, rebalanceOffsets.get(TOPIC_PARTITION3).offset());
+ sinkTaskContext.getValue().offset(offsets);
+ rebalanceListener.getValue().onPartitionsAssigned(rebalancedPartitions);
+ rebalanced.set(true);
+
+ // Run the previous async commit handler
+ asyncCallbackRunner.get().run();
+
+ // And prep the two records to return
+ long timestamp = RecordBatch.NO_TIMESTAMP;
+ TimestampType timestampType = TimestampType.NO_TIMESTAMP_TYPE;
+ List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION, FIRST_OFFSET + recordsReturnedTp1 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET + recordsReturnedTp3 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ recordsReturnedTp1 += 1;
+ recordsReturnedTp3 += 1;
+ return new ConsumerRecords<>(Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), records));
+ };
+
+ // onPartitionsRevoked
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // onPartitionsAssigned - step 1
+ final long offsetTp1 = rebalanceOffsets.get(TOPIC_PARTITION).offset();
+ final long offsetTp2 = rebalanceOffsets.get(TOPIC_PARTITION2).offset();
+ final long offsetTp3 = rebalanceOffsets.get(TOPIC_PARTITION3).offset();
+
+ // iter 4 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(postRebalanceCurrentOffsets)).thenReturn(postRebalanceCurrentOffsets);
+
+ // Setup mocks
+ when(consumer.assignment())
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions));
+
+ when(consumer.position(TOPIC_PARTITION))
+ .thenAnswer(partition1InitialPosition)
Review Comment:
nit: this can be FIRST_OFFSET
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -447,6 +457,85 @@ public void testPollRedelivery() {
assertSinkMetricValue("offset-commit-completion-total", 1.0);
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPollRedeliveryWithConsumerRebalance() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ Set<TopicPartition> newAssignment = new HashSet<>(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3));
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(newAssignment, newAssignment, newAssignment)
+ .thenReturn(Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(1))
+ // Empty consumer poll (all partitions are paused) with rebalance; one new partition is assigned
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(0))
+ // Non-empty consumer poll; all initially-assigned partitions are revoked in rebalance, and new partitions are allowed to resume
+ .thenAnswer(invocation -> {
+ ConsumerRecord<byte[], byte[]> newRecord = new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET, RAW_KEY, RAW_VALUE);
+
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptyList());
+ return new ConsumerRecords<>(Collections.singletonMap(TOPIC_PARTITION3, Collections.singletonList(newRecord)));
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // If a retriable exception is thrown, we should redeliver the same batch, pausing the consumer in the meantime
+ doThrow(new RetriableException("retry"))
+ .doThrow(new RetriableException("retry"))
+ .doThrow(new RetriableException("retry"))
+ .doThrow(new RetriableException("retry"))
Review Comment:
nit: Since put is called 5 times, and the 5th one succeeds, can you put a doNothing after all of the doThrow calls?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -642,6 +1292,228 @@ public void testTaskCancelPreventsFinalOffsetCommit() {
verify(sinkTask).close(any());
}
+ // Verify that when commitAsync is called but the supplied callback is not called by the consumer before a
+ // rebalance occurs, the async callback does not reset the last committed offset from the rebalance.
+ // See KAFKA-5731 for more information.
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCommitWithOutOfOrderCallback() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ // iter 1
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalance = invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ };
+
+ Answer<Set<TopicPartition>> initialAssignment = invocation -> INITIAL_ASSIGNMENT;
+ Answer<Long> partition1InitialPosition = invocation -> FIRST_OFFSET;
+ Answer<Long> partition2InitialPosition = invocation -> FIRST_OFFSET;
+
+ // iter 2
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPoll = expectConsumerPoll(1);
+
+ expectTaskGetTopic();
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final List<TopicPartition> originalPartitions = new ArrayList<>(INITIAL_ASSIGNMENT);
+ final List<TopicPartition> rebalancedPartitions = asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3);
+ final Map<TopicPartition, OffsetAndMetadata> rebalanceOffsets = new HashMap<>();
+ rebalanceOffsets.put(TOPIC_PARTITION, workerCurrentOffsets.get(TOPIC_PARTITION));
+ rebalanceOffsets.put(TOPIC_PARTITION2, workerCurrentOffsets.get(TOPIC_PARTITION2));
+ rebalanceOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> postRebalanceCurrentOffsets = new HashMap<>();
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 3));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET + 2));
+
+ Answer<Set<TopicPartition>> originalPartitionAssignment = invocation -> new HashSet<>(originalPartitions);
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that the callback is not called until after the rebalance
+ // changes the lastCommittedOffsets. To fake this for tests we have the commitAsync build a function
+ // that will call the callback with the appropriate parameters, and we'll run that function later.
+ final AtomicReference<Runnable> asyncCallbackRunner = new AtomicReference<>();
+ final AtomicBoolean asyncCallbackRan = new AtomicBoolean();
+
+ doAnswer(invocation -> {
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+ asyncCallbackRunner.set(() -> {
+ callback.onComplete(offsets, null);
+ asyncCallbackRan.set(true);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ // Expect the next poll to discover and perform the rebalance, THEN complete the previous callback handler,
+ // and then return one record for TP1 and one for TP3.
+ final AtomicBoolean rebalanced = new AtomicBoolean();
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalanced = invocation -> {
+ // Rebalance always begins with revoking current partitions ...
+ rebalanceListener.getValue().onPartitionsRevoked(originalPartitions);
+ // Respond to the rebalance
+ Map<TopicPartition, Long> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, rebalanceOffsets.get(TOPIC_PARTITION).offset());
+ offsets.put(TOPIC_PARTITION2, rebalanceOffsets.get(TOPIC_PARTITION2).offset());
+ offsets.put(TOPIC_PARTITION3, rebalanceOffsets.get(TOPIC_PARTITION3).offset());
+ sinkTaskContext.getValue().offset(offsets);
+ rebalanceListener.getValue().onPartitionsAssigned(rebalancedPartitions);
+ rebalanced.set(true);
+
+ // Run the previous async commit handler
+ asyncCallbackRunner.get().run();
+
+ // And prep the two records to return
+ long timestamp = RecordBatch.NO_TIMESTAMP;
+ TimestampType timestampType = TimestampType.NO_TIMESTAMP_TYPE;
+ List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION, FIRST_OFFSET + recordsReturnedTp1 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET + recordsReturnedTp3 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ recordsReturnedTp1 += 1;
+ recordsReturnedTp3 += 1;
+ return new ConsumerRecords<>(Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), records));
+ };
+
+ // onPartitionsRevoked
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // onPartitionsAssigned - step 1
+ final long offsetTp1 = rebalanceOffsets.get(TOPIC_PARTITION).offset();
+ final long offsetTp2 = rebalanceOffsets.get(TOPIC_PARTITION2).offset();
+ final long offsetTp3 = rebalanceOffsets.get(TOPIC_PARTITION3).offset();
+
+ // iter 4 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(postRebalanceCurrentOffsets)).thenReturn(postRebalanceCurrentOffsets);
+
+ // Setup mocks
+ when(consumer.assignment())
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions));
+
+ when(consumer.position(TOPIC_PARTITION))
+ .thenAnswer(partition1InitialPosition)
+ .thenReturn(offsetTp1);
+
+ when(consumer.position(TOPIC_PARTITION2))
+ .thenAnswer(partition2InitialPosition)
+ .thenReturn(offsetTp2);
+
+ when(consumer.position(TOPIC_PARTITION3))
+ .thenReturn(offsetTp3);
+
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer(consumerPollRebalance)
+ .thenAnswer(consumerPoll)
Review Comment:
nit: Inline these two calls, the `consumerPollRebalanced` one that has the big implementation in it can stay as-is.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+
+ executor.execute(() -> {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ callback.onComplete(offsets, null);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
Review Comment:
in the original test:
Wow this seems like a lot of machinery.
I think instead we can run the iteration(), capture the callback, perform the isCommitting assertion, then run the callback all on the main test thread.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -642,6 +1292,228 @@ public void testTaskCancelPreventsFinalOffsetCommit() {
verify(sinkTask).close(any());
}
+ // Verify that when commitAsync is called but the supplied callback is not called by the consumer before a
+ // rebalance occurs, the async callback does not reset the last committed offset from the rebalance.
+ // See KAFKA-5731 for more information.
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCommitWithOutOfOrderCallback() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ // iter 1
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalance = invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ };
+
+ Answer<Set<TopicPartition>> initialAssignment = invocation -> INITIAL_ASSIGNMENT;
+ Answer<Long> partition1InitialPosition = invocation -> FIRST_OFFSET;
+ Answer<Long> partition2InitialPosition = invocation -> FIRST_OFFSET;
+
+ // iter 2
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPoll = expectConsumerPoll(1);
+
+ expectTaskGetTopic();
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final List<TopicPartition> originalPartitions = new ArrayList<>(INITIAL_ASSIGNMENT);
+ final List<TopicPartition> rebalancedPartitions = asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3);
+ final Map<TopicPartition, OffsetAndMetadata> rebalanceOffsets = new HashMap<>();
+ rebalanceOffsets.put(TOPIC_PARTITION, workerCurrentOffsets.get(TOPIC_PARTITION));
+ rebalanceOffsets.put(TOPIC_PARTITION2, workerCurrentOffsets.get(TOPIC_PARTITION2));
+ rebalanceOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> postRebalanceCurrentOffsets = new HashMap<>();
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 3));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ postRebalanceCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET + 2));
+
+ Answer<Set<TopicPartition>> originalPartitionAssignment = invocation -> new HashSet<>(originalPartitions);
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that the callback is not called until after the rebalance
+ // changes the lastCommittedOffsets. To fake this for tests we have the commitAsync build a function
+ // that will call the callback with the appropriate parameters, and we'll run that function later.
+ final AtomicReference<Runnable> asyncCallbackRunner = new AtomicReference<>();
+ final AtomicBoolean asyncCallbackRan = new AtomicBoolean();
+
+ doAnswer(invocation -> {
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+ asyncCallbackRunner.set(() -> {
+ callback.onComplete(offsets, null);
+ asyncCallbackRan.set(true);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ // Expect the next poll to discover and perform the rebalance, THEN complete the previous callback handler,
+ // and then return one record for TP1 and one for TP3.
+ final AtomicBoolean rebalanced = new AtomicBoolean();
+ Answer<ConsumerRecords<byte[], byte[]>> consumerPollRebalanced = invocation -> {
+ // Rebalance always begins with revoking current partitions ...
+ rebalanceListener.getValue().onPartitionsRevoked(originalPartitions);
+ // Respond to the rebalance
+ Map<TopicPartition, Long> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, rebalanceOffsets.get(TOPIC_PARTITION).offset());
+ offsets.put(TOPIC_PARTITION2, rebalanceOffsets.get(TOPIC_PARTITION2).offset());
+ offsets.put(TOPIC_PARTITION3, rebalanceOffsets.get(TOPIC_PARTITION3).offset());
+ sinkTaskContext.getValue().offset(offsets);
+ rebalanceListener.getValue().onPartitionsAssigned(rebalancedPartitions);
+ rebalanced.set(true);
+
+ // Run the previous async commit handler
+ asyncCallbackRunner.get().run();
+
+ // And prep the two records to return
+ long timestamp = RecordBatch.NO_TIMESTAMP;
+ TimestampType timestampType = TimestampType.NO_TIMESTAMP_TYPE;
+ List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>();
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION, FIRST_OFFSET + recordsReturnedTp1 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ records.add(new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET + recordsReturnedTp3 + 1, timestamp, timestampType,
+ 0, 0, RAW_KEY, RAW_VALUE, new RecordHeaders(), Optional.empty()));
+ recordsReturnedTp1 += 1;
+ recordsReturnedTp3 += 1;
+ return new ConsumerRecords<>(Collections.singletonMap(new TopicPartition(TOPIC, PARTITION), records));
+ };
+
+ // onPartitionsRevoked
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // onPartitionsAssigned - step 1
+ final long offsetTp1 = rebalanceOffsets.get(TOPIC_PARTITION).offset();
+ final long offsetTp2 = rebalanceOffsets.get(TOPIC_PARTITION2).offset();
+ final long offsetTp3 = rebalanceOffsets.get(TOPIC_PARTITION3).offset();
+
+ // iter 4 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(postRebalanceCurrentOffsets)).thenReturn(postRebalanceCurrentOffsets);
+
+ // Setup mocks
+ when(consumer.assignment())
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(initialAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenAnswer(originalPartitionAssignment)
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions))
+ .thenReturn(new HashSet<>(rebalancedPartitions));
+
+ when(consumer.position(TOPIC_PARTITION))
+ .thenAnswer(partition1InitialPosition)
+ .thenReturn(offsetTp1);
+
+ when(consumer.position(TOPIC_PARTITION2))
+ .thenAnswer(partition2InitialPosition)
Review Comment:
nit: this can be FIRST_OFFSET
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+
+ executor.execute(() -> {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ callback.onComplete(offsets, null);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit in progress
+
+ // Make sure the "committing" flag didn't immediately get flipped back to false due to an incorrect timeout
+ assertTrue("Expected worker to be in the process of committing offsets", workerTask.isCommitting());
+
+ // Let the async commit finish and wait for it to end
+ latch.countDown();
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerCurrentOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testSinkTasksHandleCloseErrors() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(1));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ doAnswer(invocation -> null)
+ // Throw an exception on the next put to trigger shutdown behavior
+ // This exception is the true "cause" of the failure
Review Comment:
this comment is wrong
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+
+ executor.execute(() -> {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ callback.onComplete(offsets, null);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit in progress
+
+ // Make sure the "committing" flag didn't immediately get flipped back to false due to an incorrect timeout
+ assertTrue("Expected worker to be in the process of committing offsets", workerTask.isCommitting());
+
+ // Let the async commit finish and wait for it to end
+ latch.countDown();
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerCurrentOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testSinkTasksHandleCloseErrors() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(1));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ doAnswer(invocation -> null)
+ // Throw an exception on the next put to trigger shutdown behavior
+ // This exception is the true "cause" of the failure
+ .doAnswer(invocation -> {
+ workerTask.stop();
+ return null;
+ })
+ .when(sinkTask).put(anyList());
+
+ Throwable closeException = new RuntimeException();
+ when(sinkTask.preCommit(anyMap())).thenReturn(Collections.emptyMap());
+
+ // Throw another exception while closing the task's assignment
+ doThrow(closeException).when(sinkTask).close(any(Collection.class));
+
+ try {
+ workerTask.execute();
+ fail("workerTask.execute should have thrown an exception");
+ } catch (RuntimeException e) {
+ assertSame("Exception from close should propagate as-is", closeException, e);
+ }
Review Comment:
in original test:
I think this can be assertThrows
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
Review Comment:
this should probably be mentioned down below where the verification for commitSync/commitAsync would go if it existed.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
Review Comment:
nit: these warnings are also unnecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on code in PR #15316:
URL: https://github.com/apache/kafka/pull/15316#discussion_r1494825277
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+
+ executor.execute(() -> {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ callback.onComplete(offsets, null);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit in progress
+
+ // Make sure the "committing" flag didn't immediately get flipped back to false due to an incorrect timeout
+ assertTrue("Expected worker to be in the process of committing offsets", workerTask.isCommitting());
+
+ // Let the async commit finish and wait for it to end
+ latch.countDown();
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerCurrentOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testSinkTasksHandleCloseErrors() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(1));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ doAnswer(invocation -> null)
+ // Throw an exception on the next put to trigger shutdown behavior
+ // This exception is the true "cause" of the failure
Review Comment:
removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on code in PR #15316:
URL: https://github.com/apache/kafka/pull/15316#discussion_r1494861815
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -447,6 +457,85 @@ public void testPollRedelivery() {
assertSinkMetricValue("offset-commit-completion-total", 1.0);
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPollRedeliveryWithConsumerRebalance() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ Set<TopicPartition> newAssignment = new HashSet<>(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3));
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(newAssignment, newAssignment, newAssignment)
+ .thenReturn(Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3),
+ Collections.singleton(TOPIC_PARTITION3));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(1))
+ // Empty consumer poll (all partitions are paused) with rebalance; one new partition is assigned
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ .thenAnswer(expectConsumerPoll(0))
+ // Non-empty consumer poll; all initially-assigned partitions are revoked in rebalance, and new partitions are allowed to resume
+ .thenAnswer(invocation -> {
+ ConsumerRecord<byte[], byte[]> newRecord = new ConsumerRecord<>(TOPIC, PARTITION3, FIRST_OFFSET, RAW_KEY, RAW_VALUE);
+
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptyList());
+ return new ConsumerRecords<>(Collections.singletonMap(TOPIC_PARTITION3, Collections.singletonList(newRecord)));
Review Comment:
You're right, the test was a bit off mainly because the first call to `task.put(..)` shouldn't have thrown an exception.
Here's the sequence now (which matches the original WorkerSinkTaskTest):
1. Iteration#1: partitions on INITIAL_ASSIGNMENT are assigned
2. Iteration#2: `task.put(...)` throws, partitions are paused
3. Iteration#3: P3 is assigned, `task.put(...)` throws (task is already paused so it's a noop)
4. Iteration#4: `task.put(...)` throws, noop
5. Iteration#5: initial assignment is revoked (pending messages for P1 are removed); `consumer.poll(...)` returns a record for P3, which is successfully processed.
I've added the corresponding verifications after each step. Hope it makes sense now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on PR #15316:
URL: https://github.com/apache/kafka/pull/15316#issuecomment-1952966858
Thanks @gharris1727 for your review. I think I addressed most of your comments.
If anyone has any other suggestions, please let me know. Ideally we should rename this file back to `WorkerSinkTaskTest` (not sure if as part of this PR or a separate one) and have checkstyle, build.gradle, etc. cleaned up
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on code in PR #15316:
URL: https://github.com/apache/kafka/pull/15316#discussion_r1494816048
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+
+ executor.execute(() -> {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ callback.onComplete(offsets, null);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit in progress
+
+ // Make sure the "committing" flag didn't immediately get flipped back to false due to an incorrect timeout
+ assertTrue("Expected worker to be in the process of committing offsets", workerTask.isCommitting());
+
+ // Let the async commit finish and wait for it to end
+ latch.countDown();
+ executor.shutdown();
+ executor.awaitTermination(30, TimeUnit.SECONDS);
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerCurrentOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testSinkTasksHandleCloseErrors() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(1));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ doAnswer(invocation -> null)
+ // Throw an exception on the next put to trigger shutdown behavior
+ // This exception is the true "cause" of the failure
+ .doAnswer(invocation -> {
+ workerTask.stop();
+ return null;
+ })
+ .when(sinkTask).put(anyList());
+
+ Throwable closeException = new RuntimeException();
+ when(sinkTask.preCommit(anyMap())).thenReturn(Collections.emptyMap());
+
+ // Throw another exception while closing the task's assignment
+ doThrow(closeException).when(sinkTask).close(any(Collection.class));
+
+ try {
+ workerTask.execute();
+ fail("workerTask.execute should have thrown an exception");
+ } catch (RuntimeException e) {
+ assertSame("Exception from close should propagate as-is", closeException, e);
+ }
Review Comment:
I think the goal is not just to ensure that an exception is thrown, but that the exception object instance is exactly the one being set by the mock (and not some other exception)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #15316:
URL: https://github.com/apache/kafka/pull/15316#issuecomment-1938598989
As part of this last batch, please also removed this test from the list here: https://github.com/apache/kafka/blob/5cfcc52fb3fce4a43ca77df311382d7a02a40ed2/build.gradle#L424
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on PR #15316:
URL: https://github.com/apache/kafka/pull/15316#issuecomment-1940100133
Final batch of the `WorkerSinkTaskTest` migration 🎉 🎉 🎉
This one is probably the largest of the 3 (sorry).
We can either do one final commit renaming the test from `WorkerSinkTaskTestMockito` back to `WorkerSinkTastTest`, or just open a new PR once this gets merged to do the cleanup
Tagging @divijvaidya @gharris1727 @C0urante
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on code in PR #15316:
URL: https://github.com/apache/kafka/pull/15316#discussion_r1494806542
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
Review Comment:
good catch... updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14683 Migrate WorkerSinkTaskTest to Mockito (3/3) [kafka]
Posted by "hgeraldino (via GitHub)" <gi...@apache.org>.
hgeraldino commented on code in PR #15316:
URL: https://github.com/apache/kafka/pull/15316#discussion_r1494896646
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskMockitoTest.java:
##########
@@ -601,6 +690,567 @@ public void testPartialRevocationAndAssignment() {
verify(sinkTask, times(4)).put(Collections.emptyList());
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testPreCommitFailureAfterPartialRevocationAndAssignment() {
+ createTask(initialState);
+ expectTaskGetTopic();
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ when(consumer.assignment())
+ .thenReturn(INITIAL_ASSIGNMENT, INITIAL_ASSIGNMENT)
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)))
+ .thenReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION2, TOPIC_PARTITION3)));
+
+ INITIAL_ASSIGNMENT.forEach(tp -> when(consumer.position(tp)).thenReturn(FIRST_OFFSET));
+ when(consumer.position(TOPIC_PARTITION3)).thenReturn(FIRST_OFFSET);
+
+ // First poll; assignment is [TP1, TP2]
+ when(consumer.poll(any(Duration.class)))
+ .thenAnswer((Answer<ConsumerRecords<byte[], byte[]>>) invocation -> {
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ })
+ // Second poll; a single record is delivered from TP1
+ .thenAnswer(expectConsumerPoll(1))
+ // Third poll; assignment changes to [TP2]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.singleton(TOPIC_PARTITION));
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.emptySet());
+ return ConsumerRecords.empty();
+ })
+ // Fourth poll; assignment changes to [TP2, TP3]
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(Collections.emptySet());
+ rebalanceListener.getValue().onPartitionsAssigned(Collections.singleton(TOPIC_PARTITION3));
+ return ConsumerRecords.empty();
+ })
+ // Fifth poll; an offset commit takes place
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // First iteration--first call to poll, first consumer assignment
+ workerTask.iteration();
+ // Second iteration--second call to poll, delivery of one record
+ workerTask.iteration();
+ // Third iteration--third call to poll, partial consumer revocation
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+ doAnswer(invocation -> null).when(consumer).commitSync(offsets);
+
+ workerTask.iteration();
+ verify(sinkTask).close(Collections.singleton(TOPIC_PARTITION));
+ verify(sinkTask, times(2)).put(Collections.emptyList());
+
+ // Fourth iteration--fourth call to poll, partial consumer assignment
+ workerTask.iteration();
+
+ verify(sinkTask).open(Collections.singleton(TOPIC_PARTITION3));
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ workerCurrentOffsets.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ // Fifth iteration--task-requested offset commit with failure in SinkTask::preCommit
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration();
+
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION3, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testWakeupInCommitSyncCausesRetry() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ time.sleep(30000L);
+ workerTask.initializeAndStart();
+ time.sleep(30000L);
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ rebalanceListener.getValue().onPartitionsRevoked(INITIAL_ASSIGNMENT);
+ rebalanceListener.getValue().onPartitionsAssigned(INITIAL_ASSIGNMENT);
+ return ConsumerRecords.empty();
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // poll for initial assignment
+ time.sleep(30000L);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // first one raises wakeup
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.iteration(); // first record delivered
+
+ workerTask.iteration(); // now rebalance with the wakeup triggered
+ time.sleep(30000L);
+
+ verify(sinkTask).close(INITIAL_ASSIGNMENT);
+ verify(sinkTask, times(2)).open(INITIAL_ASSIGNMENT);
+
+ INITIAL_ASSIGNMENT.forEach(tp -> {
+ verify(consumer).resume(Collections.singleton(tp));
+ });
+
+ verify(statusListener).onResume(taskId);
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.33333);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 1.0);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testWakeupNotThrownDuringShutdown() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(invocation -> {
+ // stop the task during its second iteration
+ workerTask.stop();
+ return new ConsumerRecords<>(Collections.emptyMap());
+ });
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // fail the first time
+ doThrow(new WakeupException())
+ // and succeed the second time
+ .doAnswer(invocation -> null)
+ .when(consumer).commitSync(eq(offsets));
+
+ workerTask.execute();
+
+ assertEquals(0, workerTask.commitFailures());
+ verify(consumer).wakeup();
+ verify(sinkTask).close(any(Collection.class));
+ }
+
+ @Test
+ public void testRequestCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ // Initial assignment
+ time.sleep(30000L);
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+
+ final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+ offsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ offsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(offsets)).thenReturn(offsets);
+
+ // First record delivered
+ workerTask.iteration();
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 1.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.333333);
+ assertSinkMetricValue("offset-commit-seq-no", 0.0);
+ assertSinkMetricValue("offset-commit-completion-total", 0.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.5);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 0.0);
+
+ // Grab the commit time prior to requesting a commit.
+ // This time should advance slightly after committing.
+ // KAFKA-8229
+ final long previousCommitValue = workerTask.getNextCommit();
+ sinkTaskContext.getValue().requestCommit();
+ assertTrue(sinkTaskContext.getValue().isCommitRequested());
+ assertNotEquals(offsets, workerTask.lastCommittedOffsets());
+
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ time.sleep(10000L);
+ workerTask.iteration(); // triggers the commit
+ verify(consumer).commitAsync(eq(offsets), callback.capture());
+ callback.getValue().onComplete(offsets, null);
+ time.sleep(10000L);
+
+ assertFalse(sinkTaskContext.getValue().isCommitRequested()); // should have been cleared
+ assertEquals(offsets, workerTask.lastCommittedOffsets());
+ assertEquals(0, workerTask.commitFailures());
+
+ // Assert the next commit time advances slightly, the amount it advances
+ // is the normal commit time less the two sleeps since it started each
+ // of those sleeps were 10 seconds.
+ // KAFKA-8229
+ assertEquals("Should have only advanced by 40 seconds",
+ previousCommitValue +
+ (WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT - 10000L * 2),
+ workerTask.getNextCommit());
+
+ assertSinkMetricValue("partition-count", 2);
+ assertSinkMetricValue("sink-record-read-total", 1.0);
+ assertSinkMetricValue("sink-record-send-total", 1.0);
+ assertSinkMetricValue("sink-record-active-count", 0.0);
+ assertSinkMetricValue("sink-record-active-count-max", 1.0);
+ assertSinkMetricValue("sink-record-active-count-avg", 0.2);
+ assertSinkMetricValue("offset-commit-seq-no", 1.0);
+ assertSinkMetricValue("offset-commit-completion-total", 1.0);
+ assertSinkMetricValue("offset-commit-skip-total", 0.0);
+ assertTaskMetricValue("status", "running");
+ assertTaskMetricValue("running-ratio", 1.0);
+ assertTaskMetricValue("pause-ratio", 0.0);
+ assertTaskMetricValue("batch-size-max", 1.0);
+ assertTaskMetricValue("batch-size-avg", 0.33333);
+ assertTaskMetricValue("offset-commit-max-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-avg-time-ms", 0.0);
+ assertTaskMetricValue("offset-commit-failure-percentage", 0.0);
+ assertTaskMetricValue("offset-commit-success-percentage", 1.0);
+ }
+
+ @Test
+ public void testPreCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ final Map<TopicPartition, OffsetAndMetadata> taskOffsets = new HashMap<>();
+ taskOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1)); // act like FIRST_OFFSET+2 has not yet been flushed by the task
+ taskOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET + 1)); // should be ignored because > current offset
+ taskOffsets.put(new TopicPartition(TOPIC, 3), new OffsetAndMetadata(FIRST_OFFSET)); // should be ignored because this partition is not assigned
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(taskOffsets);
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> committableOffsets = new HashMap<>();
+ committableOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ committableOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerCurrentOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ // Expect extra invalid topic partition to be filtered, which causes the consumer assignment to be logged
+ ArgumentCaptor<OffsetCommitCallback> callback = ArgumentCaptor.forClass(OffsetCommitCallback.class);
+ verify(consumer).commitAsync(eq(committableOffsets), callback.capture());
+ callback.getValue().onComplete(committableOffsets, null);
+
+ assertEquals(committableOffsets, workerTask.lastCommittedOffsets());
+ }
+
+ @Test
+ public void testPreCommitFailure() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ // Put one message through the task to get some offsets to commit
+ .thenAnswer(expectConsumerPoll(2))
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ // iter 3
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 2));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenThrow(new ConnectException("Failed to flush"));
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+
+ verify(consumer).seek(TOPIC_PARTITION, FIRST_OFFSET);
+ verify(consumer).seek(TOPIC_PARTITION2, FIRST_OFFSET);
+ }
+
+ @Test
+ public void testIgnoredCommit() {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ // iter 1
+ expectPollInitialAssignment()
+ // iter 2
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerStartingOffsets);
+
+ sinkTaskContext.getValue().requestCommit();
+ workerTask.iteration(); // iter 3 -- commit
+ }
+
+ // Test that the commitTimeoutMs timestamp is correctly computed and checked in WorkerSinkTask.iteration()
+ // when there is a long running commit in process. See KAFKA-4942 for more information.
+ @Test
+ public void testLongRunningCommitWithoutTimeout() throws InterruptedException {
+ createTask(initialState);
+
+ workerTask.initialize(TASK_CONFIG);
+ workerTask.initializeAndStart();
+ verifyInitializeTask();
+
+ expectTaskGetTopic();
+ expectPollInitialAssignment()
+ .thenAnswer(expectConsumerPoll(1))
+ // no actual consumer.commit() triggered
+ .thenAnswer(expectConsumerPoll(0));
+ expectConversionAndTransformation(null, new RecordHeaders());
+
+ final Map<TopicPartition, OffsetAndMetadata> workerStartingOffsets = new HashMap<>();
+ workerStartingOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET));
+ workerStartingOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ workerTask.iteration(); // iter 1 -- initial assignment
+ assertEquals(workerStartingOffsets, workerTask.currentOffsets());
+ assertEquals(workerStartingOffsets, workerTask.lastCommittedOffsets());
+
+ time.sleep(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT);
+ workerTask.iteration(); // iter 2 -- deliver 2 records
+
+ final Map<TopicPartition, OffsetAndMetadata> workerCurrentOffsets = new HashMap<>();
+ workerCurrentOffsets.put(TOPIC_PARTITION, new OffsetAndMetadata(FIRST_OFFSET + 1));
+ workerCurrentOffsets.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
+
+ // iter 3 - note that we return the current offset to indicate they should be committed
+ when(sinkTask.preCommit(workerCurrentOffsets)).thenReturn(workerCurrentOffsets);
+
+ // We need to delay the result of trying to commit offsets to Kafka via the consumer.commitAsync
+ // method. We do this so that we can test that we do not erroneously mark a commit as timed out
+ // while it is still running and under time. To fake this for tests we have the commit run in a
+ // separate thread and wait for a latch which we control back in the main thread.
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ @SuppressWarnings("unchecked")
+ final Map<TopicPartition, OffsetAndMetadata> offsets = invocation.getArgument(0);
+ @SuppressWarnings("unchecked")
+ final OffsetCommitCallback callback = invocation.getArgument(1);
+
+ executor.execute(() -> {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+
+ callback.onComplete(offsets, null);
+ });
+
+ return null;
+ }).when(consumer).commitAsync(eq(workerCurrentOffsets), any(OffsetCommitCallback.class));
Review Comment:
Yeah, I tried to keep it as close to the original implementation as possible, but I agree this can be simplified.
Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org