storm整合kafka需要哪些包

storm整合kafka需要哪些包

答：这个是scala的编译库喝运行时库（kafka是用sbt管理依赖的），所以建议你使用sbt，会自动下载所有依赖

public partitionmanager(dynamicpartitionconnections connections, string topologyinstanceid, zkstate state, map stormconf, spoutconfig spoutconfig, partition id) { _partition = id; _connections = connections; _spoutconfig = spoutconfig; _topologyinstanceid = topologyinstanceid; _consumer = connections.register(id.host, id.partition); //注册partition到connections，并生成simpleconsumer _state = state; _stormconf = stormconf; string jsontopologyid = null; long jsonoffset = null; string path = committedpath(); try { map json = _state.readjson(path); log.info("read partition information from: " + path + " --> " + json ); if (json != null) { jsontopologyid = (string) ((map) json.get("topology")).get("id"); jsonoffset = (long) json.get("offset"); // 从zk中读出commited offset } } catch (throwable e) { log.warn("error reading and/or parsing at zknode: " + path, e); } if (jsontopologyid == null || jsonoffset == null) { // zk中没有记录，那么根据spoutconfig.startoffsettime设置offset，earliest或latest _committedto = kafkautils.getoffset(_consumer, spoutconfig.topic, id.partition, spoutconfig); log.info("no partition information found, using configuration to determine offset"); } else if (!topologyinstanceid.equals(jsontopologyid) && spoutconfig.forcefromstart) { _committedto = kafkautils.getoffset(_consumer, spoutconfig.topic, id.partition, spoutconfig.startoffsettime); log.info("topology change detected and reset from start forced, using configuration to determine offset"); } else { _committedto = jsonoffset; } _emittedtooffset = _committedto; // 初始化时，中间状态都是一致的 }