AyoubOm commented on code in PR #15361: URL: https://github.com/apache/kafka/pull/15361#discussion_r1493740130
########## streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java: ########## @@ -217,21 +215,53 @@ public void shouldFlushStateManagerWithOffsets() { final Map<TopicPartition, Long> expectedOffsets = new HashMap<>(); expectedOffsets.put(t1, 52L); expectedOffsets.put(t2, 100L); + globalStateTask.initialize(); globalStateTask.update(record(topic1, 1, 51, "foo".getBytes(), "foo".getBytes())); globalStateTask.flushState(); + assertEquals(expectedOffsets, stateMgr.changelogOffsets()); + assertTrue(stateMgr.flushed); } @Test public void shouldCheckpointOffsetsWhenStateIsFlushed() { final Map<TopicPartition, Long> expectedOffsets = new HashMap<>(); expectedOffsets.put(t1, 102L); expectedOffsets.put(t2, 100L); + globalStateTask.initialize(); globalStateTask.update(record(topic1, 1, 101, "foo".getBytes(), "foo".getBytes())); globalStateTask.flushState(); - assertThat(stateMgr.changelogOffsets(), equalTo(expectedOffsets)); + + assertEquals(expectedOffsets, stateMgr.changelogOffsets()); + assertTrue(stateMgr.checkpointWritten); + } + + @Test + public void shouldNotCheckpointIfNotReceivedEnoughRecords() { + globalStateTask.initialize(); + globalStateTask.update(record(topic1, 1, 9000L, "foo".getBytes(), "foo".getBytes())); + globalStateTask.maybeCheckpoint(); + + assertEquals(offsets, stateMgr.changelogOffsets()); + assertFalse(stateMgr.flushed); + assertFalse(stateMgr.checkpointWritten); + } + + @Test + public void shouldCheckpointIfReceivedEnoughRecords() { + final Map<TopicPartition, Long> expectedOffsets = new HashMap<>(); + expectedOffsets.put(t1, 10051L); // t1 advanced with 10.001 records Review Comment: fixed -- 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