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:
[email protected]
