ableegoldman commented on code in PR #12835:
URL: https://github.com/apache/kafka/pull/12835#discussion_r1022441487
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutorTest.java:
##########
@@ -35,4 +45,47 @@ public void testPunctuateWithPause() {
taskExecutor.punctuate();
verify(tasks).activeTasks();
}
+
+ @Test
+ public void testCommitWithOpenTransactionButNoOffsetsEOSV2() {
+ final Tasks tasks = mock(Tasks.class);
+ final TaskManager taskManager = mock(TaskManager.class);
+ final ConsumerGroupMetadata groupMetadata =
mock(ConsumerGroupMetadata.class);
+ when(taskManager.consumerGroupMetadata()).thenReturn(groupMetadata);
+
+ final TaskExecutionMetadata metadata =
mock(TaskExecutionMetadata.class);
+ final StreamsProducer producer = mock(StreamsProducer.class);
+ when(metadata.processingMode()).thenReturn(EXACTLY_ONCE_V2);
+ when(taskManager.threadProducer()).thenReturn(producer);
+ when(producer.transactionInFlight()).thenReturn(true);
+
+ final TaskExecutor taskExecutor = new TaskExecutor(tasks, taskManager,
metadata, new LogContext());
+ taskExecutor.commitOffsetsOrTransaction(Collections.emptyMap());
+
+ verify(producer).commitTransaction(Collections.emptyMap(),
groupMetadata);
+ }
+
+ @Test
+ public void testCommitWithOpenTransactionButNoOffsetsEOSV1() {
+ final TaskId taskId = new TaskId(0, 0);
+ final Task task = mock(Task.class);
+ when(task.id()).thenReturn(taskId);
Review Comment:
Nope, wasn't aware of this helper method's existence -- thanks for the tip
--
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
