nicktelford commented on code in PR #15264:
URL: https://github.com/apache/kafka/pull/15264#discussion_r1476235869
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -1073,6 +1078,35 @@ int commit(final Collection<Task> tasksToCommit) {
assertTrue(committed.get());
}
+ @Test
+ public void shouldCommitEarlyIfNeeded() {
+ final long commitInterval = 1000L;
+ final Properties props = configProps(false);
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG, stateDir);
+ props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
Long.toString(commitInterval));
+
+ final StreamsConfig config = new StreamsConfig(props);
+ final ConsumerGroupMetadata consumerGroupMetadata =
mock(ConsumerGroupMetadata.class);
+ when(consumer.groupMetadata()).thenReturn(consumerGroupMetadata);
+
when(consumerGroupMetadata.groupInstanceId()).thenReturn(Optional.empty());
+ final Task runningTask = mock(Task.class);
+ final TaskManager taskManager = mockTaskManagerCommit(runningTask, 2);
+ when(taskManager.needsCommit(anyBoolean())).thenReturn(false);
+
+ final TopologyMetadata topologyMetadata = new
TopologyMetadata(internalTopologyBuilder, config);
+ topologyMetadata.buildAndRewriteTopology();
+ thread = buildStreamThread(consumer, taskManager, config,
topologyMetadata);
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+
+ when(taskManager.needsCommit(anyBoolean())).thenReturn(true);
+ mockTime.sleep(commitInterval - 10L);
+ thread.setNow(mockTime.milliseconds());
+ thread.maybeCommit();
+
+ verify(taskManager, times(2)).commit(mkSet(runningTask));
Review Comment:
I've added the third case, and I've independently verified the invocations
of `commit`. I've used `Mockito.clearInvocations` to ensure that the
invocations I'm verifying are linked to the correct case; if you know of a
better way to do this, please let me know.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
