我们有一个小的微服务，可以使用Spring Cloud Stream从Kafka主题读取并写入mqtt。它工作正常，但一段时间后，我们得到以下异常，并且没有进一步的消息发布到mqtt：

"2022-10-18 16:22:29.861 WARN 1 --- [d | tellus-mqtt] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-mqtt-2, groupId=mqtt] consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches r

是否有一种方法可以通过编程方式重新订阅或恢复此超时？
我们是否可以为执行器实现一个自定义的健康检查，以包括消费者，然后pod将自动重新启动k8s？类似于：

management:
  endpoint:
    health:
      group:
        liveness:
          include: livenessstate,binders

其中 binders 是Kafka组件。
编辑：以下是消费者代码（OutputConfig类）：

@Configuration
@Log4j2
@Profile("output")
public class OutputConfig {

    private final Mqtt3ReactorClient outboundMqttClient;
    private final Mqtt3ReactorClient outboundRootMqttClient;
    private final MeterUtils meterUtils;

    @Autowired
    public OutputConfig(@Qualifier("outboundMqttClient") Mqtt3ReactorClient outboundMqttClient,
                        @Qualifier("outboundRootMqttClient") Mqtt3ReactorClient outboundRootMqttClient,
                        MeterUtils meterUtils) {
        this.outboundMqttClient = outboundMqttClient;
        this.outboundRootMqttClient = outboundRootMqttClient;
        this.meterUtils = meterUtils;
        log.info("Starting Output Config!");
    }

    @Bean
    public Consumer<Flux<Output.GatewayNotification>> kafka() {
        return new Output(outboundMqttClient, meterUtils);
    }

    @Bean
    public Consumer<Flux<Output.GatewayNotification>> kafkaRoot() {
        return new Output(outboundRootMqttClient, meterUtils);
    }

}

和Output类：

@Log4j2
public class Output implements Consumer<Flux<Output.GatewayNotification>> {

    public static final HexFormat FORMAT = HexFormat.of().withDelimiter(" ").withUpperCase();
    private final Mqtt3ReactorClient outboundMqttClient;
    private final MeterUtils meterUtils;

    public Output(Mqtt3ReactorClient outboundMqttClient, MeterUtils meterUtils) {
        this.outboundMqttClient = outboundMqttClient;
        this.meterUtils = meterUtils;
    }

    @Override
    public void accept(Flux<Output.GatewayNotification> gatewayNotifications) {
        Flux<Mqtt3Publish> messagesToPublish = gatewayNotifications
                .map(gatewayNotification -> Mqtt3Publish.builder()
                        .topic(gatewayNotification.getAddress())
                        .qos(MqttQos.AT_LEAST_ONCE)
                        .payload(Base64.getDecoder().decode(gatewayNotification.getPayload()))
                        .build());

        outboundMqttClient.publish(messagesToPublish)
                .doOnNext(publishResult -> {
                    log.debug(
                            "Publish acknowledged: " + FORMAT.formatHex(publishResult.getPublish().getPayloadAsBytes()));
                    meterUtils.incrementCounter("output");
                })
                .doOnError(error -> log.error(error.getMessage()))
                .subscribe();
    }

    @Data
    public static class GatewayNotification {
        private String address;
        private String payload;
        private Long buildingId;
    }

HiveMqMqttConfig：

@Configuration
@Log4j2
public class HiveMqMqttConfig {

    @Value("${mqtt.endpointUrl}")
    private String endpointUrl;

    @Value("${mqtt.rootEndpointUrl}")
    private String rootEndpointUrl;

    @Value("${mqtt.inboundClientId}")
    private String inboundClientId;

    @Value("${mqtt.outboundClientId}")
    private String outboundClientId;

    @Value("${mqtt.caFilename:#{null}}")
    private String caFilename;

    @Value("${mqtt.inboundPrivateKeyFilename:#{null}}")
    private String inboundPrivateKeyFilename;

    @Value("${mqtt.inboundRootPrivateKeyFilename:#{null}}")
    private String inboundRootPrivateKeyFilename;

    @Value("${mqtt.inboundClientCertFilename:#{null}}")
    private String inboundClientCertFilename;

    @Value("${mqtt.inboundRootClientCertFilename:#{null}}")
    private String inboundRootClientCertFilename;

    @Value("${mqtt.outboundPrivateKeyFilename:#{null}}")
    private String outboundPrivateKeyFilename;

    @Value("${mqtt.outboundRootPrivateKeyFilename:#{null}}")
    private String outboundRootPrivateKeyFilename;

    @Value("${mqtt.outboundClientCertFilename:#{null}}")
    private String outboundClientCertFilename;

    @Value("${mqtt.outboundRootClientCertFilename:#{null}}")
    private String outboundRootClientCertFilename;

    @Bean(name = "inboundMqttClient")
    public Mqtt3ReactorClient inboundMqttClient() {
        var client = Mqtt3ReactorClient.from(buildMqtt3Client(endpointUrl, UUID.randomUUID().toString(), caFilename, inboundPrivateKeyFilename, inboundClientCertFilename));
        connectClient(client);
        return client;
    }

    @Bean(name = "inboundRootMqttClient")
    public Mqtt3ReactorClient inboundRootMqttClient() {
        var client = Mqtt3ReactorClient.from(buildMqtt3Client(rootEndpointUrl, UUID.randomUUID().toString(), caFilename, inboundRootPrivateKeyFilename, inboundRootClientCertFilename));
        connectClient(client);
        return client;
    }

    @Bean(name = "outboundMqttClient")
    public Mqtt3ReactorClient outboundMqttClient() {
        var client = Mqtt3ReactorClient.from(buildMqtt3Client(endpointUrl, UUID.randomUUID().toString(), caFilename, outboundPrivateKeyFilename, outboundClientCertFilename));
        connectClient(client);
        return client;
    }

    @Bean(name = "outboundRootMqttClient")
    public Mqtt3ReactorClient outboundRootMqttClient() {
        var client = Mqtt3ReactorClient.from(buildMqtt3Client(rootEndpointUrl, UUID.randomUUID().toString(), caFilename, outboundRootPrivateKeyFilename, outboundRootClientCertFilename));
        connectClient(client);
        return client;
    }

    private Mqtt3Client buildMqtt3Client(String endpointUrl, String clientId, String caFilename, String privateKeyFilename, String clientCertFilename) {
        log.info("Creating mqtt3 client with client id: {}", clientId);
        // endpoint is in the form 'protocol://host:port'
        String[] endpointUrlComponents = endpointUrl.split(":");
        String host = endpointUrlComponents[1].substring(2);
        int port = Integer.parseInt(endpointUrlComponents[2]);

        Mqtt3ClientBuilder mqtt3ClientBuilder = Mqtt3Client.builder()
                .identifier(clientId)
                .serverHost(host)
                .serverPort(port)
                .automaticReconnectWithDefaultConfig();

        try {
            if (caFilename != null && !caFilename.isEmpty()) {
                boolean isUsingKeyBasedAuthentication = privateKeyFilename != null && !privateKeyFilename.isEmpty() && clientCertFilename != null && !clientCertFilename.isEmpty();

                PemFileSslContext context
                        = isUsingKeyBasedAuthentication
                        ? new PemFileSslContext(getStreamFromClassPathOrLocal(caFilename), getStreamFromClassPathOrLocal(privateKeyFilename), getStreamFromClassPathOrLocal(clientCertFilename))
                        : new PemFileSslContext(new ClassPathResource(caFilename).getInputStream());

                context.getSocketFactory();
                mqtt3ClientBuilder
                        .sslConfig()
                        .keyManagerFactory(context.getKeyManagerFactory())
                        .trustManagerFactory(context.getTrustManagerFactory())
                        .applySslConfig();

            }
        } catch (IOException | NoSuchAlgorithmException | KeyStoreException | CertificateException |
                 InvalidKeySpecException | UnrecoverableKeyException | PemFileSslContext.SocketFactoryCreationFailedException e) {
            throw new RuntimeException(e);
        }

        return mqtt3ClientBuilder.build();
    }

    private InputStream getStreamFromClassPathOrLocal(String uri) throws IOException {
        return new ClassPathResource(uri).getInputStream();
    }

    private void connectClient(Mqtt3ReactorClient mqtt3ReactorClient) {
        Mono<Mqtt3ConnAck> connAckSingle = mqtt3ReactorClient.connect();

        connAckSingle
                .doOnSuccess(connAck -> log.info("Connected, " + connAck.getReturnCode()))
                .doOnError(throwable -> log.info("Connection failed, " + throwable.getMessage()))
                .subscribe();
    }
}

配置：

management:
  endpoint:
    health:
      group:
        liveness:
          include: livenessstate,kafkaConsumers

spring:
  cloud:
    stream:
      kafka:
        bindings:
          kafka-in-0:
            consumer:
              configuration:
                max.poll.records: 10
          kafkaRoot-in-0:
            consumer:
              configuration:
                max.poll.records: 10
      function:
        definition: kafka;kafkaRoot
      bindings:
        kafka-in-0:
          destination: output
          group: mqtt
          consumer:
            concurrency: 1
        kafkaRoot-in-0:
          destination: output
          group: mqtt-root
          consumer:
            concurrency: 1

... (certs/endpoints omitted)