如何通过Java代码实现Kafka Streams应用重置？

可以实现，Kafka官方提供了原生Java API可以完成和kafka-streams-application-reset.sh完全等价的操作，无需调用外部Shell命令。

前提注意 #

• 该操作仅适用于开发测试环境，禁止在生产环境默认开启自动重置逻辑，会导致业务数据重复处理或丢失
• 执行重置前必须确保所有使用对应application.id的Kafka Streams实例已完全停止，否则重置操作会失效

实现步骤 #

1. 引入依赖 #

首先确保项目中引入了Kafka Streams依赖，以Maven为例：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>匹配你的Kafka集群版本即可</version>
</dependency>

操作Topic需要的kafka-client依赖一般会由kafka-streams传递引入，无需额外单独配置。

2. 核心重置代码 #

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import java.util.Properties;

public class KafkaStreamsResetUtil {
    public static void resetStreamsApp(String bootstrapServers, String applicationId, Topology businessTopology) {
        // 构造和业务流应用完全一致的配置，如有安全、序列化等配置也可同步传入
        Properties props = new Properties();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);

        // 此处传入业务实际的Topology即可，仅执行cleanUp操作不需要启动流
        KafkaStreams streams = new KafkaStreams(businessTopology, props);
        // 清理本地状态存储
        streams.cleanUp();
        streams.close();
    }
}

3. 扩展：批量删除中间Topic、重置消费组偏移 #

如果要完全匹配官方脚本的全部能力，还可以配合AdminClient完成额外操作：

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import java.util.List;
import java.util.Properties;
import java.util.Collections;
import java.util.concurrent.ExecutionException;

public void extraResetOperations(String bootstrapServers, String applicationId, List<String> intermediateTopics) throws ExecutionException, InterruptedException {
    Properties adminProps = new Properties();
    adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    try (Admin admin = Admin.create(adminProps)) {
        // *注意：需要Kafka集群开启delete.topic.enable=true配置才能成功删除Topic
        admin.deleteTopics(intermediateTopics).all().get();
        // 重置对应消费组的偏移量到最早位置
        admin.resetOffsets(Collections.singleton(applicationId), OffsetResetStrategy.EARLIEST).all().get();
    }
}

使用建议 #

• 给重置逻辑加环境开关，仅在开发环境启动时才触发执行
• 多应用场景下可以把上述逻辑封装为公共工具类，循环传入所有需要重置的application.id和对应中间Topic列表即可，无需重复写代码

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