@Test public void testMaybeClean() throws Exception { final File baseDir = Files.createTempDirectory("test").toFile(); try { final long cleanupDelay = 1000L; final Properties props = configProps(false); props.setProperty(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG, Long.toString(cleanupDelay)); props.setProperty(StreamsConfig.STATE_DIR_CONFIG, baseDir.getCanonicalPath()); final StreamsConfig config = new StreamsConfig(props); final File applicationDir = new File(baseDir, applicationId); applicationDir.mkdir(); final File stateDir1 = new File(applicationDir, task1.toString()); final File stateDir2 = new File(applicationDir, task2.toString()); final File stateDir3 = new File(applicationDir, task3.toString()); final File extraDir = new File(applicationDir, applicationId); stateDir1.mkdir(); stateDir2.mkdir(); stateDir3.mkdir(); extraDir.mkdir(); 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 maybeClean(final long now) { super.maybeClean(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); assertTrue(thread.tasks().isEmpty()); mockTime.sleep(cleanupDelay); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertTrue(stateDir3.exists()); assertTrue(extraDir.exists()); List<TopicPartition> revokedPartitions; List<TopicPartition> assignedPartitions; Map<TaskId, StreamTask> prevTasks; final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>(); activeTasks.put(task1, Collections.singleton(t1p1)); activeTasks.put(task2, Collections.singleton(t1p2)); thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks)); revokedPartitions = Collections.emptyList(); assignedPartitions = Arrays.asList(t1p1, t1p2); prevTasks = new HashMap<>(thread.tasks()); final ConsumerRebalanceListener rebalanceListener = thread.rebalanceListener; rebalanceListener.onPartitionsRevoked(revokedPartitions); rebalanceListener.onPartitionsAssigned(assignedPartitions); assertTrue(prevTasks.isEmpty()); assertEquals(2, thread.tasks().size()); mockTime.sleep(cleanupDelay - 10L); thread.maybeClean(mockTime.milliseconds()); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertTrue(stateDir3.exists()); assertTrue(extraDir.exists()); mockTime.sleep(11L); thread.maybeClean(mockTime.milliseconds()); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertFalse(stateDir3.exists()); assertTrue(extraDir.exists()); activeTasks.clear(); revokedPartitions = assignedPartitions; assignedPartitions = Collections.emptyList(); prevTasks = new HashMap<>(thread.tasks()); rebalanceListener.onPartitionsRevoked(revokedPartitions); rebalanceListener.onPartitionsAssigned(assignedPartitions); assertEquals(2, prevTasks.size()); for (final StreamTask task : prevTasks.values()) { assertTrue(((TestStreamTask) task).committed); ((TestStreamTask) task).committed = false; } assertTrue(thread.tasks().isEmpty()); mockTime.sleep(cleanupDelay - 10L); thread.maybeClean(mockTime.milliseconds()); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertFalse(stateDir3.exists()); assertTrue(extraDir.exists()); mockTime.sleep(11L); thread.maybeClean(mockTime.milliseconds()); assertFalse(stateDir1.exists()); assertFalse(stateDir2.exists()); assertFalse(stateDir3.exists()); assertTrue(extraDir.exists()); } finally { Utils.delete(baseDir); } }
protected void maybeClean(final long now) { if (now > lastCleanMs + cleanTimeMs) { stateDirectory.cleanRemovedTasks(cleanTimeMs); lastCleanMs = now; } }
StreamThread extends Thread { protected void maybeClean(final long now) { if (now > lastCleanMs + cleanTimeMs) { stateDirectory.cleanRemovedTasks(cleanTimeMs); lastCleanMs = now; } } }
StreamThread extends Thread { protected void maybeClean(final long now) { if (now > lastCleanMs + cleanTimeMs) { stateDirectory.cleanRemovedTasks(cleanTimeMs); lastCleanMs = now; } } 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 maybeClean(final long now) { if (now > lastCleanMs + cleanTimeMs) { stateDirectory.cleanRemovedTasks(cleanTimeMs); lastCleanMs = now; } } 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 maybeClean(final long now) { if (now > lastCleanMs + cleanTimeMs) { stateDirectory.cleanRemovedTasks(cleanTimeMs); lastCleanMs = now; } } 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; }