我正在运行cloudera提供的kafka版本0.9.0,我已经编写了一个定制的生产者
我正在将以下属性添加到producerconfig:
Properties props = new Properties();
props.put("bootstrap.servers", brokers);
props.put("acks", "all");
//props.put("metadata.broker.list", this.getBrokers());
//props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("partitioner.class", "com.vikas.MyPartitioner");
然而,当我运行我的程序,我得到这个错误
16/08/23 18:00:33 INFO producer.ProducerConfig: ProducerConfig values:
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
key.serializer = class org.apache.kafka.common.serialization.StringSerializer
block.on.buffer.full = true
retry.backoff.ms = 100
buffer.memory = 33554432
batch.size = 16384
metrics.sample.window.ms = 30000
metadata.max.age.ms = 300000
receive.buffer.bytes = 32768
timeout.ms = 30000
max.in.flight.requests.per.connection = 5
bootstrap.servers = [server01:9092, server02:9092]
metric.reporters = []
client.id =
compression.type = none
retries = 0
max.request.size = 1048576
send.buffer.bytes = 131072
acks = all
reconnect.backoff.ms = 10
linger.ms = 1
metrics.num.samples = 2
metadata.fetch.timeout.ms = 60000
16/08/23 18:00:33 WARN producer.ProducerConfig: The configuration partitioner.class = null was supplied but isn't a known config.
根据文件http://kafka.apache.org/090/documentation.html#producerconfigs
分区器,类是一个有效的属性,但我不知道为什么Kafka抱怨它是一个未知的配置。
1条答案
按热度按时间p4tfgftt1#
看看这篇文章http://www.javaworld.com/article/3066873/big-data/big-data-messaging-with-kafka-part-2.html 它提供了如何使用自定义分区器的示例。
如您所见,错误消息是
“16/08/23 18:00:33 warn producer.producerconfig:提供了配置partitioner.class=null,但不是已知配置。”这意味着kafka不理解密钥partitioner.class
你可以这样设置分区器