Tag: 汇合

KafkaAvroDeserializer不返回SpecificRecord但返回GenericRecord: 我的KafkaProducer能够使用KafkaAvroSerializer将对象序列化为我的主题。但是， KafkaConsumer.poll()返回反序列化的GenericRecord而不是我的序列化类。 MyKafkaProducer KafkaProducer producer; try (InputStream props = Resources.getResource(“producer.props”).openStream()) { Properties properties = new Properties(); properties.load(props); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroSerializer.class); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroSerializer.class); properties.put(“schema.registry.url”, “http://localhost:8081”); MyBean bean = new MyBean(); producer = new KafkaProducer(properties); producer.send(new ProducerRecord(topic, bean.getId(), bean)); 我的KafkaConsumer try (InputStream props = Resources.getResource(“consumer.props”).openStream()) { properties.load(props); properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class); properties.put(“schema.registry.url”, “http://localhost:8081”); consumer = new KafkaConsumer(properties); […]

从本地计算机连接到Docker中运行的Kafka: 我在本地计算机上使用docker设置单节点基本Kafka部署，如Confluent Kafka文档中所述（步骤2-3）。另外，我还暴露了zookeeper的端口2181和kafka的端口9092，这样我就能从本地机器上运行的java客户端连接到它们： $ docker run -d \ -p 2181:2181 \ –net=confluent \ –name=zookeeper \ -e ZOOKEEPER_CLIENT_PORT=2181 \ confluentinc/cp-zookeeper:4.1.0 $ docker run -d \ –net=confluent \ –name=kafka \ -p 9092:9092 \ -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \ -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \ -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \ confluentinc/cp-kafka:4.1.0 问题：当我尝试从主机连接到kafka时，连接失败，因为它无法解析地址：kafka：9092。这是我的Java代码： Properties props = new Properties(); props.put(“bootstrap.servers”, “localhost:9092”); props.put(“client.id”, “KafkaExampleProducer”); props.put(“key.serializer”, LongSerializer.class.getName()); […]