带有Kafka Binder的Spring Cloud Stream： /bindings执行器API不会停止生产者

Question

我有一个带有Actuator和Kafka活页夹的Spring Cloud Stream项目。我正在探索绑定/执行器，并试图停止生产者作为一个练习。我通过curl提出以下POST请求：

curl-v'localhost:8081/actuator/bindings/producer-out-0'-H'内容类型：应用程序/json'-d'{"state"："STOPPED"}'

实际结果：查询返回204。生产者的状态（从GET /actuator/bindings/producer-out-0看到）现在是停止。然而，生产者仍在产生消息，这可以从日志记录和消费者关于该主题的活动中看到。

预期结果：我希望制作人停止生成消息。（我还尝试使用暂停状态，该状态也返回204，但错误日志表明此生产者无法暂停。）

我是不是误解了这个执行器的工作原理？当一个制作人被停止时，是否期望S. C. S.会继续对该制作人进行投票？我知道的唯一留档在这里，但据我所知，它没有回答我的问题。

背景：

我使用的是SpringBootStarter父版本2.5.3，并将StarterWeb和StarterActuator列为依赖项。我想我没有遗漏任何东西。

这是生产者/消费者对。正如你所看到的，我使用的是一个可调查的供应商。

@Configuration
@Profile("numbers")
public class NumberHandlers {
  private static final Logger LOGGER = LoggerFactory.getLogger(NumberHandlers.class);

  @Bean
  public Supplier<Integer> producer() {
    // Needed an effectively-final mutable integer. Side-bar comments welcome. :P
    var counter = new AtomicInteger();
    return () -> {
      var n = counter.getAndIncrement();
      LOGGER.info("Producing number: " + n);
      return n;
    };
  }

  @Bean
  public Consumer<Integer> consumer() {
    return it -> LOGGER.info("Consuming number: " + it);
  }
}

当我传入编号配置文件时，这些都处于活动状态。我的配置如下。

application.yml：

server:
  port: 8081
spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: ${env.kafka.bootstrapservers:localhost}
management:
  endpoints:
    web:
      exposure:
        include: 'bindings'

...application-numbers.yml：

spring:
  cloud:
    stream:
      poller:
        fixedDelay: 5000
      bindings:
        producer-out-0:
          destination: numbers-raw
          producer:
            partitionCount: 3
        consumer-in-0:
          destination: numbers-raw
      kafka:
        bindings:
          producer-out-0:
            producer:
              topic.properties:
                # These look weird because they're done as an exercise.
                retention.bytes: 10000
                retention.ms: 172800000
    function:
      definition: producer;consumer

我正在本地主机环境中使用docker compose kafka和zookeeper在主机网络上进行测试。

谢谢！

Answer 1

当前不支持生产者绑定的生命周期控制，仅支持使用者绑定。