Windows环境下Kafka Streams TopologyTestDriver关闭失败求助

我之前也踩过这个Windows专属的坑，跟Kafka Streams 2.3.0版本在Windows系统下未及时释放状态存储的文件句柄直接相关。针对你遇到的这个问题，这里有几个可行的解决方案：

1. 升级Kafka Streams到2.4.0及以上版本（最推荐） #

这个问题其实是官方已知的bug，已经在2.4.0版本中被修复了。官方调整了StateDirectory清理逻辑，针对Windows文件系统的特性优化了文件句柄的释放时机，升级后就能彻底解决这个异常。如果你的项目允许升级依赖版本，这是最省心的解决方案。

2. 手动指定状态目录并强制删除（临时Workaround） #

如果暂时没法升级版本，可以给测试指定一个专属的临时状态目录，在关闭TopologyTestDriver后手动递归删除目录，绕过Windows的文件句柄限制：

首先修改你的getKafkaProperties方法，添加状态目录配置：

private static Properties getKafkaProperties() {
    Properties properties = new Properties();
    properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-app");
    properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "");
    // 指定专属临时目录，避免和其他测试冲突
    properties.put(StreamsConfig.STATE_DIR_CONFIG, "./tmp/kafka-streams-test");
    return properties;
}

然后修改tearDown方法，在关闭驱动后手动删除目录：

@AfterEach
void tearDown() {
    if (testDriver != null) {
        testDriver.close();
        // 强制删除状态目录
        Path stateDir = Paths.get("./tmp/kafka-streams-test");
        try {
            Files.walkFileTree(stateDir, new SimpleFileVisitor<Path>() {
                @Override
                public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
                    Files.delete(file);
                    return FileVisitResult.CONTINUE;
                }

                @Override
                public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                    Files.delete(dir);
                    return FileVisitResult.CONTINUE;
                }
            });
        } catch (IOException e) {
            // 测试环境下可以记录日志或忽略
            e.printStackTrace();
        }
    }
}

3. 临时切换处理语义（应急用，不推荐） #

你的测试中使用了EXACTLY_ONCE处理语义，这个模式下Kafka Streams会创建更多的状态文件和句柄，临时切换为AT_LEAST_ONCE可能会规避这个问题，但这会改变测试的语义，只适合应急场景：

properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.AT_LEAST_ONCE);

问题根源解释 #

Windows和Unix类系统的文件系统行为差异导致了这个问题：Unix允许删除被打开的文件（文件会被标记为删除，直到所有句柄关闭才真正清理），但Windows不允许删除仍有打开句柄的文件或目录。Kafka Streams 2.3.0的TopologyTestDriver在处理聚合类操作（比如groupByKey+reduce）时，RocksDB状态存储的文件句柄可能未在关闭驱动时及时释放，导致StateDirectory清理目录时抛出DirectoryNotEmptyException。

内容的提问来源于stack exchange，提问作者philipp94831