在Spring boot中暂停/启动Kafka流处理器

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
</dependency>

@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public KafkaStreamsConfiguration kafkaStreamsConfiguration() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, groupId);
        // 其他配置属性...

        return new KafkaStreamsConfiguration(props);
    }
}

@Component
public class KafkaStreamProcessor {

    private KafkaStreams kafkaStreams;

    @Autowired
    public KafkaStreamProcessor(KafkaStreamsConfiguration kafkaStreamsConfiguration) {
        StreamsBuilder builder = new StreamsBuilder();
        // 定义流处理逻辑...

        kafkaStreams = new KafkaStreams(builder.build(), kafkaStreamsConfiguration.asProperties());
    }

    public void start() {
        kafkaStreams.start();
    }

    public void pause() {
        kafkaStreams.pause();
    }

    public void resume() {
        kafkaStreams.resume();
    }
}

@RestController
public class KafkaStreamController {

    @Autowired
    private KafkaStreamProcessor kafkaStreamProcessor;

    @PostMapping("/pause")
    public void pauseStreamProcessor() {
        kafkaStreamProcessor.pause();
    }

    @PostMapping("/resume")
    public void resumeStreamProcessor() {
        kafkaStreamProcessor.resume();
    }
}