我在 Windows 上使用 Confluent 3.0.1 平台。我按照安装指南和开发人员指南进行所有安装和开发我的拓扑。
我启动了 Zookeeper,然后启动了 Kafka 服务器并尝试运行我的拓扑。但是在 Kafka 服务器上出现以下错误。即使我手动创建主题并运行拓扑,我也会看到相同的错误。
INFO Topic creation {"version":1,"partitions":{"0":[0]}} (kafka.admin.AdminUtils$)
[2016-09-21 17:20:08,807] INFO [KafkaApi-0] Auto creation of topic Text4 with 1 partitions and replication factor 1 is successful (kafka.server.KafkaApis)
[2016-09-21 17:20:09,436] ERROR [KafkaApi-0] Error when handling request {group_id=my-first-streams-application1} (kafka.server.KafkaApis)
kafka.admin.AdminOperationException: replication factor: 3 larger than available brokers: 1
at kafka.admin.AdminUtils$.assignReplicasToBrokers(AdminUtils.scala:117)
at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:403)
at kafka.server.KafkaApis.kafka$server$KafkaApis$$createTopic(KafkaApis.scala:629)
at kafka.server.KafkaApis.kafka$server$KafkaApis$$createGroupMetadataTopic(KafkaApis.scala:651)
at kafka.server.KafkaApis$$anonfun$getOrCreateGroupMetadataTopic$1.apply(KafkaApis.scala:657)
at kafka.server.KafkaApis$$anonfun$getOrCreateGroupMetadataTopic$1.apply(KafkaApis.scala:657)
at scala.Option.getOrElse(Option.scala:121)
at kafka.server.KafkaApis.getOrCreateGroupMetadataTopic(KafkaApis.scala:657)
at kafka.server.KafkaApis.handleGroupCoordinatorRequest(KafkaApis.scala:818)
at kafka.server.KafkaApis.handle(KafkaApis.scala:86)
at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:60)
at java.lang.Thread.run(Thread.java:745)
我的拓扑代码如下:
public class MetricTopology implements InitializingBean {
@Autowired
@Qualifier("getStreamsConfig")
private Properties properties;
// Method to build topology.
public void buildTopology() {
System.out.println("MetricTopology.buildTopology()");
TopologyBuilder builder = new TopologyBuilder();
// add the source processor node that takes Kafka topic "Text4" as input
builder.addSource("Source", "Text4")
// add the Metricsprocessor node which takes the source processor as its upstream processor
.addProcessor("Process", () -> new MetricsProcessor(), "Source");
// Building Stream.
KafkaStreams streams = new KafkaStreams(builder, properties);
streams.start();
}
// Called after all properties are set.
public void afterPropertiesSet() throws Exception {
buildTopology();
}
}
以下是我正在使用的属性,它是不同 java 源文件的一部分。
Properties settings = new Properties();
// Set a few key parameters. This properties will be picked from property file.
settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-first-streams-application1");
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
settings.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, "1");