StreamThread extends Thread { protected void maybeCommit() { long now = time.milliseconds(); if (commitTimeMs >= 0 && lastCommit + commitTimeMs < now) { log.trace("Committing processor instances because the commit interval has elapsed."); commitAll(); lastCommit = now; processStandbyRecords = true; } } StreamThread(TopologyBuilder builder,
                        StreamsConfig config,
                        KafkaClientSupplier clientSupplier,
                        String applicationId,
                        String clientId,
                        UUID processId,
                        Metrics metrics,
                        Time time); void partitionAssignor(StreamPartitionAssignor partitionAssignor); @Override void run(); void close(); Map<TaskId, StreamTask> tasks(); Set<TaskId> prevTasks(); Set<TaskId> cachedTasks(); final PartitionGrouper partitionGrouper; final String applicationId; final String clientId; final UUID processId; }
@Test public void testMaybeCommit() throws Exception { File baseDir = Files.createTempDirectory("test").toFile(); try { final long commitInterval = 1000L; Properties props = configProps(); props.setProperty(StreamsConfig.STATE_DIR_CONFIG, baseDir.getCanonicalPath()); props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.toString(commitInterval)); StreamsConfig config = new StreamsConfig(props); MockTime mockTime = new MockTime(); TopologyBuilder builder = new TopologyBuilder(); builder.addSource("source1", "topic1"); StreamThread thread = new StreamThread(builder, config, new MockClientSupplier(), applicationId, clientId, processId, new Metrics(), mockTime) { @Override public void maybeCommit() { super.maybeCommit(); } @Override protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitionsForTask) { ProcessorTopology topology = builder.build("X", id.topicGroupId); return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer, producer, restoreConsumer, config); } }; initPartitionGrouper(config, thread); ConsumerRebalanceListener rebalanceListener = thread.rebalanceListener; List<TopicPartition> revokedPartitions; List<TopicPartition> assignedPartitions; revokedPartitions = Collections.emptyList(); assignedPartitions = Arrays.asList(t1p1, t1p2); rebalanceListener.onPartitionsRevoked(revokedPartitions); rebalanceListener.onPartitionsAssigned(assignedPartitions); assertEquals(2, thread.tasks().size()); mockTime.sleep(commitInterval - 10L); thread.maybeCommit(); for (StreamTask task : thread.tasks().values()) { assertFalse(((TestStreamTask) task).committed); } mockTime.sleep(11L); thread.maybeCommit(); for (StreamTask task : thread.tasks().values()) { assertTrue(((TestStreamTask) task).committed); ((TestStreamTask) task).committed = false; } mockTime.sleep(commitInterval - 10L); thread.maybeCommit(); for (StreamTask task : thread.tasks().values()) { assertFalse(((TestStreamTask) task).committed); } mockTime.sleep(11L); thread.maybeCommit(); for (StreamTask task : thread.tasks().values()) { assertTrue(((TestStreamTask) task).committed); ((TestStreamTask) task).committed = false; } } finally { Utils.delete(baseDir); } }