java Kafka consumer error由于节点-1断开连接,取消了相关id为1的正在进行的API_VERSIONS请求

vmpqdwk3  于 2023-06-20  发布在  Java
关注(0)|答案(4)|浏览(165)

我的消费者配置如下

package com.example.kafka.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2181");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-tenent1-id");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String>
                factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

}

但是,当应用程序启动时,我看到以下内容继续输出
由于节点断开连接,相关ID为1的正在进行的API_VERSIONS请求已取消
我可以使用以下代码向Kafka主题发送消息

kafkaTemplate.send("test-topic", msg);

使用者侦听器如下所示

@Service
public class Receiver {
    @KafkaListener(topics = "test-topic", groupId = "group-tenent1-id")
    public void listen(String message) {
        log.info("Received Messasge in group - group-id: " + message);
    }
}
package com.example.kafka.config;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

但我无法记录收到的消息

5rgfhyps

5rgfhyps1#

在我的情况下,我使用AWS MSK引导服务器与IAM。我没有为security.protocolsasl.mechanism添加适当的安全配置,在更改它之后,我的应用程序工作没有任何问题。
以前是像下面

security.protocol: PLAINTEXT
  sasl.mechanism: PLAIN

然后我把它改成了

security.protocol: SASL_SSL
  sasl.mechanism: AWS_MSK_IAM
zphenhs4

zphenhs42#

删除了Java中的所有自定义Kafka配置并将其放在application中。yml解决了这个问题。我现在可以发送和接收消息了。

server:
  port: 8080
spring:
  kafka:
    consumer:
      bootstrap-servers: localhost:9092
      group-id: group-tenant1-id
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      bootstrap-servers: localhost:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
dauxcl2d

dauxcl2d3#

您还没有显示生产者配置,但我假设它使用localhost:9092(如果它工作的话)。
您的消费者不使用此产品。
理想情况下,您应该将配置外部化到Spring属性文件中,并使用一个位置来设置spring.kafka.bootstrap-servers=localhost:9092,然后应用程序中的消费者和生产者客户端都将使用该位置。

g6ll5ycj

g6ll5ycj4#

我在Confluent Cloud Kafka上收到了这个错误消息,但在我的本地Kafka集群上没有。上述解决方案均不适用。我怀疑安全设置,但我的application.yml包含了所有的安全设置-而且正确。最终发现修复方法是将Confluent的安全设置放在Kafka Consumer Config中,如下所示(您的安全设置可能会有所不同):

kafkaConfigProperties.put("ssl.endpoint.identification.algorithm", "https"); 
kafkaConfigProperties.put("sasl.mechanism", "PLAIN"); 
kafkaConfigProperties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + apiKey + "\" password=\"" + apiPassword + "\";"); 
kafkaConfigProperties.put("security.protocol", "SASL_SSL");

相关问题