我创建了一个带有3个分区和3个重复的Kafka主题
$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test
并写了Kafka-Producer程序从API获取数据，它工作正常 . 我的数据成功存储在3个分区和3个复制中 .
现在我必须处理分区Kafka日志中的数据 . 我有一个Storm拓扑 . 我的拓扑如何接受日志中的数据作为输入 .
最初我创建了一个带有1个分区的主题，1个复制，我的风暴拓扑工作正常 . 但是当我将分区保持为3并且复制为3时，数据不会进入Storm-Topology . 如何解决它 .
我的拓扑代码
public static void main(String[] args) throws Exception{
Map map = Maps.newHashMap();
map.put("dataSourceClassName", "org.postgresql.ds.PGSimpleDataSource");
map.put("dataSource.url","jdbc:postgresql://localhost:5432/test?user=postgres");
ConnectionProvider cp = new MyConnectionProvider(map);
String argument = args[0];
Config conf = new Config();
conf.put(JDBC_CONF, map);
conf.setDebug(true);
conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 3);
//set the number of workers
conf.setNumWorkers(3);
TopologyBuilder builder = new TopologyBuilder();
//Setup Kafka spout
BrokerHosts hosts = new ZkHosts("localhost:2181");
String topic = "test";
String zkRoot = "";
String consumerGroupId = "group1";
SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);
spoutConfig.scheme = new RawMultiScheme();
spoutConfig.scheme = new SchemeAsMultiScheme(new Search_Parser());
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
builder.setSpout("KafkaSpout", kafkaSpout);
Help Me on This Issue
提前致谢