Kafka 生产者无法正常工作，出现错误 kafka-producer-network-thread |生产者-1 [重复]答案

【问题标题】：Kafka producer not working getting an error kafka-producer-network-thread | producer-1 [duplicate]Kafka 生产者无法正常工作，出现错误 kafka-producer-network-thread |生产者-1 [重复]
【发布时间】：2021-03-22 17:37:24
【问题描述】：

我的 zookeeper 和 Kafka 服务器正在运行，但我不知道为什么下面的代码不起作用。

我以前用Kafka-python用python运行Kafka，但是幂等性有问题，所以改用java。我只想用 java 测试运行基本的生产者代码，但不幸的是它不起作用。我是java新手，请帮助我。

这是我的基本生产者代码：

package Main;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class producerDemo {

    public static void main(String[] args) {

        //Create producer property
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


        //Create the producer
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //Producer create record
        ProducerRecord<String, String> record = new ProducerRecord<>("first_topic", "First message with java kafka");

        //Send data - asynchronous
        producer.send(record);

        producer.flush();
        producer.close();
    }
}

控制台日志：

[main] INFO org.apache.kafka.clients.producer.ProducerConfig - ProducerConfig values: 
    acks = 1
    batch.size = 16384
    bootstrap.servers = [127.0.0.1:9092]
    buffer.memory = 33554432
    client.dns.lookup = use_all_dns_ips
    client.id = producer-1
    compression.type = none
    connections.max.idle.ms = 540000
    delivery.timeout.ms = 120000
    enable.idempotence = false
    interceptor.classes = []
    internal.auto.downgrade.txn.commit = false
    key.serializer = class org.apache.kafka.common.serialization.StringSerializer
    linger.ms = 0
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metadata.max.idle.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2147483647
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.3
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class org.apache.kafka.common.serialization.StringSerializer

[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.6.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: 62abe01bee039651
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1607678031008
[kafka-producer-network-thread | producer-1] ERROR org.apache.kafka.common.utils.KafkaThread - Uncaught exception in thread 'kafka-producer-network-thread | producer-1':
java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/JsonNode
    at org.apache.kafka.common.requests.ApiVersionsRequest$Builder.<clinit>(ApiVersionsRequest.java:36)
    at org.apache.kafka.clients.NetworkClient.handleConnections(NetworkClient.java:910)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:555)
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:325)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:240)
    at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.lang.ClassNotFoundException: com.fasterxml.jackson.databind.JsonNode
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:606)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:168)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
    ... 6 more
[main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.

Process finished with exit code 0

Pom xml：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>java_python_Idea_01</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>15</maven.compiler.source>
        <maven.compiler.target>15</maven.compiler.target>
    </properties>

    <dependencies>

        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.6.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-simple -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.30</version>
<!--            <scope>test</scope>-->
        </dependency>


    </dependencies>

</project>

【问题讨论】：

问题是你没有必需的库。您是如何创建项目的？你用过 maven、intellij 的想法吗？
你必须传递字节数据而不是字符串。将字符串转换为字节。
@VladislavVarslavans 我正在使用 maven
请在问题中添加pom.xml 内容。
请查看这个问题：stackoverflow.com/questions/64374511/…

标签： java apache-kafka

【解决方案1】：

在@nehaev 的帮助下（谢谢）来源：NoClassDefFoundError when trying to create partitions in Kafka

ClassNotFoundException: com.fasterxml.jackson.databind.JsonNode

所以你需要

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.8.10</version>
</dependency>

但是，您确实应该使用 jackson-bom 一致地导入所有相关的依赖版本

【讨论】：

您也可以降级 Kafka 客户端，因为这是 2.6.0 中的错误