StreamThread extends Thread { protected void maybeClean() { long now = time.milliseconds(); if (now > lastClean + cleanTimeMs) { File[] stateDirs = stateDir.listFiles(); if (stateDirs != null) { for (File dir : stateDirs) { try { String dirName = dir.getName(); TaskId id = TaskId.parse(dirName.substring(dirName.lastIndexOf("-") + 1)); if (dir.exists()) { FileLock directoryLock = null; try { directoryLock = ProcessorStateManager.lockStateDirectory(dir); if (directoryLock != null) { log.info("Deleting obsolete state directory {} for task {} after delayed {} ms.", dir.getAbsolutePath(), id, cleanTimeMs); Utils.delete(dir); } } catch (FileNotFoundException e) { } catch (IOException e) { log.error("Failed to lock the state directory due to an unexpected exception", e); } finally { if (directoryLock != null) { try { directoryLock.release(); directoryLock.channel().close(); } catch (IOException e) { log.error("Failed to release the state directory lock"); } } } } } catch (TaskIdFormatException e) { } } } lastClean = now; } } 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 testMaybeClean() throws Exception { File baseDir = Files.createTempDirectory("test").toFile(); try { final long cleanupDelay = 1000L; Properties props = configProps(); props.setProperty(StreamsConfig.STATE_CLEANUP_DELAY_MS_CONFIG, Long.toString(cleanupDelay)); props.setProperty(StreamsConfig.STATE_DIR_CONFIG, baseDir.getCanonicalPath()); StreamsConfig config = new StreamsConfig(props); File applicationDir = new File(baseDir, applicationId); applicationDir.mkdir(); File stateDir1 = new File(applicationDir, task1.toString()); File stateDir2 = new File(applicationDir, task2.toString()); File stateDir3 = new File(applicationDir, task3.toString()); File extraDir = new File(applicationDir, "X"); stateDir1.mkdir(); stateDir2.mkdir(); stateDir3.mkdir(); extraDir.mkdir(); 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 maybeClean() { super.maybeClean(); } @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; 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<Integer, StreamTask> prevTasks; revokedPartitions = Collections.emptyList(); assignedPartitions = Arrays.asList(t1p1, t1p2); prevTasks = new HashMap(thread.tasks()); rebalanceListener.onPartitionsRevoked(revokedPartitions); rebalanceListener.onPartitionsAssigned(assignedPartitions); assertTrue(prevTasks.isEmpty()); assertEquals(2, thread.tasks().size()); mockTime.sleep(cleanupDelay - 10L); thread.maybeClean(); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertTrue(stateDir3.exists()); assertTrue(extraDir.exists()); mockTime.sleep(11L); thread.maybeClean(); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertFalse(stateDir3.exists()); assertTrue(extraDir.exists()); revokedPartitions = assignedPartitions; assignedPartitions = Collections.emptyList(); prevTasks = new HashMap(thread.tasks()); rebalanceListener.onPartitionsRevoked(revokedPartitions); rebalanceListener.onPartitionsAssigned(assignedPartitions); assertEquals(2, prevTasks.size()); for (StreamTask task : prevTasks.values()) { assertTrue(((TestStreamTask) task).committed); ((TestStreamTask) task).committed = false; } assertTrue(thread.tasks().isEmpty()); mockTime.sleep(cleanupDelay - 10L); thread.maybeClean(); assertTrue(stateDir1.exists()); assertTrue(stateDir2.exists()); assertFalse(stateDir3.exists()); assertTrue(extraDir.exists()); mockTime.sleep(11L); thread.maybeClean(); assertFalse(stateDir1.exists()); assertFalse(stateDir2.exists()); assertFalse(stateDir3.exists()); assertTrue(extraDir.exists()); } finally { Utils.delete(baseDir); } }