apache kafka分区
apache kafka分区
提问于 2018-07-03 12:30:38
我对apache kafka非常陌生，我正在尝试做kafka分区，同时将字符串发送给kafka，
public class Producer {
public static void main(String[] argv)throws Exception {
String topicName = "test";
//Configure the Producer
Properties configProperties = new Properties();
configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.4.226:9092");
configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");
configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
configProperties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CountryPartitioner.class.getCanonicalName());
configProperties.put("partitions.0","USA");
configProperties.put("partitions.1","India");
org.apache.kafka.clients.producer.Producer producer = new KafkaProducer(configProperties);
ProducerRecord<String, String> rec = new ProducerRecord<String, String>(topicName,"message111");
producer.send(rec, new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
System.out.println("Message sent to topic ->" + metadata.topic() +" stored at offset->" + metadata.offset());
}
});
producer.close();
}
}现在我得到了这样的例外
2018-07-03 17:52:12 ERROR RecordBatch:102 - Error executing user-provided callback on message for topic-partition test-2:
java.lang.NullPointerException
at com.spnotes.kafka.partition.Producer$1.onCompletion(Producer.java:29)
at org.apache.kafka.clients.producer.internals.RecordBatch.done(RecordBatch.java:99)
at org.apache.kafka.clients.producer.internals.RecordBatch.maybeExpire(RecordBatch.java:136)
at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortExpiredBatches(RecordAccumulator.java:220)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:192)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:141)
at java.lang.Thread.run(Thread.java:748)我做错什么了？提前谢谢。
回答 1
Stack Overflow用户
回答已采纳
发布于 2018-07-04 04:46:33
编辑3:只需创建多个分区，就可以看到这些steps.First，我们为每个代理创建一个配置文件(在Windows上使用copy命令)
> cp config/server.properties config/server-1.properties
> cp config/server.properties config/server-2.properties现在编辑这些新文件并设置以下属性：
config/server-1.properties:
broker.id=1
listeners=PLAINTEXT://:9093
log.dir=/tmp/kafka-logs-1
config/server-2.properties:
broker.id=2
listeners=PLAINTEXT://:9094
log.dir=/tmp/kafka-logs-2我们已经有了动物园管理员，并且我们的单个节点已经启动，所以我们只需要启动两个新节点：
> bin/kafka-server-start.sh config/server-1.properties &
> bin/kafka-server-start.sh config/server-2.properties &现在创建一个新主题，其复制因子为3，以及3个分区，将解决这个问题：
> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic my-replicated-topic页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/51154710
复制相关文章
相似问题
腾讯云开发者
