Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateTaskTest.java (Java)

Code before:
        globalStateTask.initialize();
        globalStateTask.update(record(topic1, 1, currentOffsetT1 + 1, "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, currentOffsetT1 + 51L, "foo".getBytes(), "foo".getBytes()));
        globalStateTask.flushState();

        assertEquals(expectedOffsets, stateMgr.changelogOffsets());
        assertTrue(stateMgr.checkpointWritten);
    }

    @Test
    public void shouldNotCheckpointIfNotReceivedEnoughRecords() {
        globalStateTask.initialize();
        // Reset after initialization since checkpointing should happen during initialization, KAFKA-18168
        stateMgr.checkpointWritten = false;
        stateMgr.flushed = false;

        globalStateTask.update(record(topic1, 1, currentOffsetT1 + 9000L, "foo".getBytes(), "foo".getBytes()));
        time.sleep(flushInterval); // flush interval elapsed
        globalStateTask.maybeCheckpoint();

        assertEquals(offsets, stateMgr.changelogOffsets());
        assertFalse(stateMgr.flushed);
        assertFalse(stateMgr.checkpointWritten);
    }

    @Test
    public void shouldNotCheckpointWhenFlushIntervalHasNotLapsed() {
        globalStateTask.initialize();
        // Reset after initialization since checkpointing should happen during initialization, KAFKA-18168
        stateMgr.checkpointWritten = false;
        stateMgr.flushed = false;

        // offset delta exceeded
        globalStateTask.update(record(topic1, 1, currentOffsetT1 + 10000L, "foo".getBytes(), "foo".getBytes()));

        time.sleep(flushInterval / 2);
        globalStateTask.maybeCheckpoint();


Review comment (suggestion):
```suggestion
        // Reset after initialization since checkpointing should happen during initialization
```

Code after:
        globalStateTask.initialize();
        globalStateTask.update(record(topic1, 1, currentOffsetT1 + 1, "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, currentOffsetT1 + 51L, "foo".getBytes(), "foo".getBytes()));
        globalStateTask.flushState();

        assertEquals(expectedOffsets, stateMgr.changelogOffsets());
        assertTrue(stateMgr.checkpointWritten);
    }

    @Test
    public void shouldNotCheckpointIfNotReceivedEnoughRecords() {
        globalStateTask.initialize();
        globalStateTask.update(record(topic1, 1, currentOffsetT1 + 9000L, "foo".getBytes(), "foo".getBytes()));
        time.sleep(flushInterval); // flush interval elapsed

        stateMgr.checkpointWritten = false;
        stateMgr.flushed = false;

        globalStateTask.maybeCheckpoint();

        assertEquals(offsets, stateMgr.changelogOffsets());
        assertFalse(stateMgr.flushed);
        assertFalse(stateMgr.checkpointWritten);
    }

    @Test
    public void shouldNotCheckpointWhenFlushIntervalHasNotLapsed() {
        globalStateTask.initialize();

        // offset delta exceeded
        globalStateTask.update(record(topic1, 1, currentOffsetT1 + 10000L, "foo".getBytes(), "foo".getBytes()));

        time.sleep(flushInterval / 2);

        stateMgr.checkpointWritten = false;
        stateMgr.flushed = false;

        globalStateTask.maybeCheckpoint();