我使用的是Storm版本1.1.0和Kafka版本0.10.1.2。
我正在创建 Kafka-spout,如下所示:
public KafkaSpout<String, String> getKafkaSpout() {
String _kafkaBrokers = (String) props.get("bootstrap.servers");
String _topic = (String) props.get("kafka.topic.name");
String groupId = (String) props.get("group.id");
int maxMsgSize = (int) props.get("fetch.message.max.bytes");
String keySerializer = (String) props.get("key.serializer");
String valueSerializer = (String) props.get("value.serializer");
List<String>topics = new ArrayList<String>(`enter code here`);
topics.add(_topic);
return new KafkaSpout<String, String (KafkaSpoutConfig.builder(_kafkaBrokers, topics)
.setFirstPollOffsetStrategy(FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST)
.setMaxUncommittedOffsets(100)
.setProp(ConsumerConfig.GROUP_ID_CONFIG, groupId)
.setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,maxMsgSize)
.setProp("key.serializer",keySerializer)
.setProp("value.serializer",valueSerializer)
.build())
}
我收到下面提到的错误
java.lang.IllegalStateException: Consumer is not subscribed to any topics or assigned any partitions
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:973)
at org.apache.storm.kafka.spout.KafkaSpout.pollKafkaBroker(KafkaSpout.java:291)
at org.apache.storm.kafka.spout.KafkaSpout.nextTuple(KafkaSpout.java:225)
at org.apache.storm.daemon.executor$fn__9798$fn__9813$fn__9844.invoke(executor.clj:647)
at org.apache.storm.util$async_loop$fn__555.invoke(util.clj:484)
at clojure.lang.AFn.run(AFn.java:22) at java.lang.Thread.run(Thread.java:745)
连同其他依赖项,我在下面提到了项目中的 Maven 依赖项
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>1.1.0.2.6.2.0-205</version>
</dependency>
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka</artifactId>
<version>1.1.0.2.6.2.0-205</version>
</dependency>
我认为
List<String>topics = new ArrayList<String>("enter code here");
是你的问题?您可能需要在该列表中写下您的主题名称。
您的依赖版本很奇怪,据我所知 Storm 尚未发布任何带有这些版本字符串的内容。
我还想知道为什么你需要 Storm-kafka-client(适用于 Kafka > 0.10 集群)和storm-kafka(适用于较旧的 Kafka 集群)(但我认为目前仍与最新的 Kafka 兼容)。