@Test public void testMaybeCommit() throws Exception { final File baseDir = Files.createTempDirectory("test").toFile(); try { final long commitInterval = 1000L; final Properties props = configProps(false); props.setProperty(StreamsConfig.STATE_DIR_CONFIG, baseDir.getCanonicalPath()); props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.toString(commitInterval)); final StreamsConfig config = new StreamsConfig(props); builder.addSource("source1", "topic1"); final StreamThread thread = new StreamThread( builder, config, clientSupplier, applicationId, clientId, processId, metrics, mockTime, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST), 0) { @Override public void maybeCommit(final long now) { super.maybeCommit(now); } @Override protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitionsForTask) { final ProcessorTopology topology = builder.build(id.topicGroupId); return new TestStreamTask( id, applicationId, partitionsForTask, topology, consumer, clientSupplier.getProducer(new HashMap<String, Object>()), restoreConsumer, config, new MockStreamsMetrics(new Metrics()), stateDirectory); } }; initPartitionGrouper(config, thread, clientSupplier); final ConsumerRebalanceListener rebalanceListener = thread.rebalanceListener; final List<TopicPartition> revokedPartitions; final 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(mockTime.milliseconds()); for (final StreamTask task : thread.tasks().values()) { assertFalse(((TestStreamTask) task).committed); } mockTime.sleep(11L); thread.maybeCommit(mockTime.milliseconds()); for (final StreamTask task : thread.tasks().values()) { assertTrue(((TestStreamTask) task).committed); ((TestStreamTask) task).committed = false; } mockTime.sleep(commitInterval - 10L); thread.maybeCommit(mockTime.milliseconds()); for (final StreamTask task : thread.tasks().values()) { assertFalse(((TestStreamTask) task).committed); } mockTime.sleep(11L); thread.maybeCommit(mockTime.milliseconds()); for (final StreamTask task : thread.tasks().values()) { assertTrue(((TestStreamTask) task).committed); ((TestStreamTask) task).committed = false; } } finally { Utils.delete(baseDir); } }
protected void maybeCommit(final long now) { if (commitTimeMs >= 0 && lastCommitMs + commitTimeMs < now) { log.debug("{} Committing all active tasks {} and standby tasks {} because the commit interval {}ms has elapsed by {}ms", logPrefix, activeTasks.keySet(), standbyTasks.keySet(), commitTimeMs, now - lastCommitMs); commitAll(); lastCommitMs = now; processStandbyRecords = true; } }
StreamThread extends Thread { protected void maybeCommit(final long now) { if (commitTimeMs >= 0 && lastCommitMs + commitTimeMs < now) { log.debug("{} Committing all active tasks {} and standby tasks {} because the commit interval {}ms has elapsed by {}ms", logPrefix, activeTasks.keySet(), standbyTasks.keySet(), commitTimeMs, now - lastCommitMs); commitAll(); lastCommitMs = now; processStandbyRecords = true; } } }
StreamThread extends Thread { protected void maybeCommit(final long now) { if (commitTimeMs >= 0 && lastCommitMs + commitTimeMs < now) { log.debug("{} Committing all active tasks {} and standby tasks {} because the commit interval {}ms has elapsed by {}ms", logPrefix, activeTasks.keySet(), standbyTasks.keySet(), commitTimeMs, now - lastCommitMs); commitAll(); lastCommitMs = now; processStandbyRecords = true; } } StreamThread(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier,
                        final String applicationId,
                        final String clientId,
                        final UUID processId,
                        final Metrics metrics,
                        final Time time,
                        final StreamsMetadataState streamsMetadataState,
                        final long cacheSizeBytes); }
StreamThread extends Thread { protected void maybeCommit(final long now) { if (commitTimeMs >= 0 && lastCommitMs + commitTimeMs < now) { log.debug("{} Committing all active tasks {} and standby tasks {} because the commit interval {}ms has elapsed by {}ms", logPrefix, activeTasks.keySet(), standbyTasks.keySet(), commitTimeMs, now - lastCommitMs); commitAll(); lastCommitMs = now; processStandbyRecords = true; } } StreamThread(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier,
                        final String applicationId,
                        final String clientId,
                        final UUID processId,
                        final Metrics metrics,
                        final Time time,
                        final StreamsMetadataState streamsMetadataState,
                        final long cacheSizeBytes); @Override void run(); synchronized void close(); synchronized boolean isInitialized(); synchronized boolean stillRunning(); Map<TaskId, StreamTask> tasks(); Set<TaskId> prevActiveTasks(); Set<TaskId> cachedTasks(); void setStateListener(final StreamThread.StateListener listener); synchronized State state(); @Override String toString(); String toString(final String indent); }
StreamThread extends Thread { protected void maybeCommit(final long now) { if (commitTimeMs >= 0 && lastCommitMs + commitTimeMs < now) { log.debug("{} Committing all active tasks {} and standby tasks {} because the commit interval {}ms has elapsed by {}ms", logPrefix, activeTasks.keySet(), standbyTasks.keySet(), commitTimeMs, now - lastCommitMs); commitAll(); lastCommitMs = now; processStandbyRecords = true; } } StreamThread(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier,
                        final String applicationId,
                        final String clientId,
                        final UUID processId,
                        final Metrics metrics,
                        final Time time,
                        final StreamsMetadataState streamsMetadataState,
                        final long cacheSizeBytes); @Override void run(); synchronized void close(); synchronized boolean isInitialized(); synchronized boolean stillRunning(); Map<TaskId, StreamTask> tasks(); Set<TaskId> prevActiveTasks(); Set<TaskId> cachedTasks(); void setStateListener(final StreamThread.StateListener listener); synchronized State state(); @Override String toString(); String toString(final String indent); final String applicationId; final String clientId; final UUID processId; }