[GitHub] [kafka] guozhangwang commented on a change in pull request #8964: KAFKA-9450: Decouple flushing state from commiting

[GitBox]

guozhangwang commented on a change in pull request #8964:
URL: https://github.com/apache/kafka/pull/8964#discussion_r468799847



##########
File path: 
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
##########
@@ -1458,50 +1498,91 @@ public void shouldThrowIfPostCommittingOnIllegalState() 
{
 
         task.transitionTo(Task.State.SUSPENDED);
         task.transitionTo(Task.State.CLOSED);
-        assertThrows(IllegalStateException.class, task::postCommit);
+        assertThrows(IllegalStateException.class, () -> task.postCommit(true));
     }
 
     @Test
     public void shouldSkipCheckpointingSuspendedCreatedTask() {
-        stateManager.checkpoint(EasyMock.anyObject());
+        stateManager.checkpoint();
         EasyMock.expectLastCall().andThrow(new AssertionError("Should not have 
tried to checkpoint"));
         EasyMock.replay(stateManager);
 
         task = createStatefulTask(createConfig(false, "100"), true);
         task.suspend();
-        task.postCommit();
+        task.postCommit(true);
     }
 
     @Test
-    public void shouldCheckpointWithEmptyOffsetsForSuspendedRestoringTask() {
-        stateManager.checkpoint(emptyMap());
+    public void shouldCheckpointForSuspendedTask() {
+        stateManager.checkpoint();
+        EasyMock.expectLastCall().once();
+        EasyMock.expect(stateManager.changelogOffsets())
+                .andReturn(Collections.singletonMap(partition1, 0L))
+                .andReturn(Collections.singletonMap(partition1, 1L));
         EasyMock.replay(stateManager);
 
         task = createStatefulTask(createConfig(false, "100"), true);
         task.initializeIfNeeded();
         task.suspend();
-        task.postCommit();
+        task.postCommit(true);
         EasyMock.verify(stateManager);
     }
 
     @Test
-    public void 
shouldCheckpointWithEmptyOffsetsForSuspendedRunningTaskWithNoCommitNeeded() {
-        stateManager.checkpoint(emptyMap());
+    public void shouldNotCheckpointForSuspendedRunningTaskWithSmallProgress() {
+        EasyMock.expect(stateManager.changelogOffsets())
+                .andReturn(Collections.singletonMap(partition1, 1L))
+                .andReturn(Collections.singletonMap(partition1, 2L))
+                .andReturn(Collections.singletonMap(partition1, 3L));
+        stateManager.checkpoint();
+        EasyMock.expectLastCall().andThrow(new AssertionError("Checkpoint 
should not be called")).anyTimes();

Review comment:
       Actually if I do not setup, it would not fail.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org