@SuppressWarnings("unchecked")
@Test
public void testMaybePunctuate() throws Exception {

    File baseDir = Files.createTempDirectory("test").toFile();
    try {
        StreamsConfig config = createConfig(baseDir);
        StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null);
        task.addRecords(partition1, records(
                new ConsumerRecord<>(partition1.topic(), partition1.partition(), 20, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
                new ConsumerRecord<>(partition1.topic(), partition1.partition(), 30, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
                new ConsumerRecord<>(partition1.topic(), partition1.partition(), 40, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue)
        ));
        task.addRecords(partition2, records(
                new ConsumerRecord<>(partition2.topic(), partition2.partition(), 25, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
                new ConsumerRecord<>(partition2.topic(), partition2.partition(), 35, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue),
                new ConsumerRecord<>(partition2.topic(), partition2.partition(), 45, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue)
        ));
        assertTrue(task.maybePunctuate());
        assertEquals(5, task.process());
        assertEquals(1, source1.numReceived);
        assertEquals(0, source2.numReceived);
        assertFalse(task.maybePunctuate());
        assertEquals(4, task.process());
        assertEquals(1, source1.numReceived);
        assertEquals(1, source2.numReceived);
        assertTrue(task.maybePunctuate());
        assertEquals(3, task.process());
        assertEquals(2, source1.numReceived);
        assertEquals(1, source2.numReceived);
        assertFalse(task.maybePunctuate());
        assertEquals(2, task.process());
        assertEquals(2, source1.numReceived);
        assertEquals(2, source2.numReceived);
        assertTrue(task.maybePunctuate());
        assertEquals(1, task.process());
        assertEquals(3, source1.numReceived);
        assertEquals(2, source2.numReceived);
        assertFalse(task.maybePunctuate());
        assertEquals(0, task.process());
        assertEquals(3, source1.numReceived);
        assertEquals(3, source2.numReceived);
        assertFalse(task.maybePunctuate());
        processor.supplier.checkAndClearPunctuateResult(20L, 30L, 40L);
        task.close();
    } finally {
        Utils.delete(baseDir);
    }
}