Kafka 2.9.1 生产者 0.8.2.1 编译与运行时依赖关系

Question

因此，在 api 0.8.2 中，生产者的 Kakfa 配置属性发生了变化；在完成该工作并让我的 java Producer 编译之后，我得到了一个异常。生产者针对我的 Kafka_2.9.1-0.8.2.1 集群的节点，我得到关于 DefaultSerializer 未实例化的异常：

Exception in thread "main" org.apache.kafka.common.KafkaException: Could not instantiate class kafka.serializer.DefaultEncoder Does it have a public no-argument constructor?
        at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:235)
        at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:136)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:216)
........

鉴于这是在 Kakfa 中实现的，我想知道将 Kafka 作为依赖项进行编译是否还不够，因为我可能需要在运行时打包到一个或多个 Kafka jar 中。我还没有找到关于此的文档（最新的或其他的）。是否缺少生产者运行时 jar？

作为参考，我将我的 build.gradle 放在这里（有点乱）。编译中的 excludes 是在已经收到此错误后新添加的，因此无论是否在 dependencies 块中有这些行，都会发生错误。我确实尝试过只依赖 0.8.2 的 kafka-client 模块，但我认为这对制作人不起作用。这是文件：

buildscript {
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath 'com.google.protobuf:protobuf-gradle-plugin:0.7.0'
    }
}

group 'lamblin'
version '0.1-SNAPSHOT'

apply plugin: 'application'
apply plugin: "com.google.protobuf"

sourceCompatibility = 1.7
targetCompatibility = 1.7

// Eliminates bootstrap class warning from javac
//tasks.withType(Compile) {
//    options.bootClasspath = "$JDK6_HOME/jre/lib/rt.jar"
//}

repositories {
    mavenCentral()
}

dependencies {
    testCompile group: 'junit', name: 'junit', version: '4.11'
    compile group: 'com.google.guava', name: 'guava', version: '18.0'
    compile group: 'com.google.protobuf', name: 'protobuf-java-util', version: '3.0.0-beta-1'
    compile group: 'com.google.transit', name: 'gtfs-realtime-bindings', version: '0.0.4'
    compile group: 'com.offbytwo', name: 'docopt', version: '0.6.0.20150202'
    //compile group: 'org.apache.kafka', name: 'kafka_2.9.1', version: '0.8.2.1' {
    compile ('org.apache.kafka:kafka_2.9.1:0.8.2.1') {
        exclude group: 'com.sun.jmx', module: 'jmxri'
        exclude group: 'javax.jmx', module: 'jms'
        exclude group: 'com.sun.jdmk', module: 'jmxtools'
    }
}

protobuf {
    generateProtoTasks {
        all().each { task ->
            task.builtins {
                python { }
            }
        }
    }
    protoc {
        //  artifact = 'com.google.protobuf:protoc:3.0.0-alpha-3'
        artifact = 'com.google.protobuf:protoc:2.6.1'
    }
}
// First Application Script
mainClassName = "com.insight.lamblin.GtfsToJson"
applicationName = "gtfsToJson"

// Subsequent Scripts
task createAllStartScripts() << {
    // This task is added to by a loop over the scripts array creating-sub tasks
}
def scripts = [ 'gtfsToJson': 'com.insight.lamblin.GtfsToJson',
                'rawGtfsKafkaProducer': 'com.insight.lamblin.RawGtfsKafkaProducer'
]
scripts.each() { scriptName, className ->
    def t = tasks.create(name: scriptName+'StartScript', type: CreateStartScripts) {
        mainClassName = className
        applicationName = scriptName
        outputDir = new File(project.buildDir, 'scripts')
        classpath = jar.outputs.files + project.configurations.runtime
    }
    applicationDistribution.into("bin") {
        from(t)
        fileMode = 0755
    }
    createAllStartScripts.dependsOn(t)
}

Answer 1

场景：在一个不起眼的教堂地下室里，一圈金属折叠椅上坐着形形色色的中年男人和几个女人，大多数人都戴着眼镜，一副冷漠的样子。一盒咖啡和一些半个甜甜圈依次放在塑料盘子上，塑料盘子放在靠近入口的墙上，上面有缺口、未覆盖、折叠 table。

Daniel：嗨，我叫 Daniel，我是...（抽泣）...文档浏览器。
小组（慢慢地）：欢迎丹尼尔。

写那个剧本是因为我的 Kafka 问题似乎只吸引了蟋蟀，所以我在这里让它有点有趣……感觉很寂寞。

在我看来，关于生产者属性设置的看似权威的 kafka.apache.org 文档有 10 多份。 kafka.serializer.DefaultSerializer 在几乎所有设置属性的示例中都非常突出和常见，而 Java 生产者示例完全没有属性或运行示例代码的详细信息。

此外，尽管名称为 "default"，但此属性没有默认值，因此需要设置它。这似乎是一个愚蠢的细节，但对于 Kafka 开发团队的某个人来说一定是有意义的。

当运行一个 Kafka 生产者用 Java 编写时，生产者应该从少数可用的 Java 特定编码器中设置编码器。前面提到的似乎是 Scala 特定的。对于 Java，您对 org.apache.kafka.common.serialization and you the equivalent to the default serializer is: ByteArraySerializer. If you set key.serializer and value.serializer it should work. The better way to set those is using the static strings in ProducerConfig 感兴趣，例如 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG。

设置有点像：

import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.ProducerConfig; 
import org.apache.kafka.clients.producer.ProducerRecord; 
... 
import java.util.Properties;
...
Properties props = new Properties(); 
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
          "172.31.22.7:9092,172.31.22.6:9092,172.31.22.5:9092,172.31.22.4:9092"); 
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
          "org.apache.kafka.common.serialization.ByteArraySerializer"); 
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
          "org.apache.kafka.common.serialization.ByteArraySerializer"); 
KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(props); 
...

Kafka 2.9.1 生产者 0.8.2.1 编译与运行时依赖关系

Kafka 2.9.1 producer 0.8.2.1 compile vs runtime dependencies

java

apache-kafka

kafka-producer-api