I dont know, i'm using BaseRichBolt.
2015-05-21 11:25 GMT-03:00 Benjamin Cuthbert <cuthbert....@gmail.com>:
> I am using BaseBasicBolt as that was told ack events
> automatically. Should we not use that?
>
>
>
> On 21 May 2015, at 15:24, Cristian Makoto Sandiga <cmsand...@gmail.com>
> wrote:
>
> Are you acking in your bolt? collector.ack(input);
>
> 2015-05-21 11:19 GMT-03:00 Benjamin Cuthbert <cuthbert....@gmail.com>:
>
>> Okay so interesting:
>>
>> I have your configuration in my local eclipse and a remote zookeeper that
>> kafka is connected to:
>>
>> ZkHosts zkhost = new ZkHosts(“1.1.1.1:2181","/brokers"); // /brokers ->
>> kafka broker
>> SpoutConfig spoutconfig = new SpoutConfig(“1.1.1.1:2181", “/prices",
>> "/brokers", “rawWarehousePriceSpout");
>>
>> Zookeeper:
>>
>> [zk: 127.0.0.1:2181(CONNECTED) 80] ls /brokers
>> [rawWarehousePriceSpout, topics, ids]
>> [zk: 127.0.0.1:2181(CONNECTED) 81] ls /brokers/rawWarehousePriceSpout
>> []
>> [zk: 127.0.0.1:2181(CONNECTED) 82]
>>
>>
>> Logs i see:
>>
>> 41994 [Thread-15-rawWarehousePriceSpout] INFO storm.kafka.KafkaUtils -
>> Task [1/1] assigned
>> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>> partition=0},
>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>> partition=1},
>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>> partition=2}]
>> 41994 [Thread-15-rawWarehousePriceSpout] INFO storm.kafka.ZkCoordinator
>> - Task [1/1] Deleted partition managers: []
>> 41994 [Thread-15-rawWarehousePriceSpout] INFO storm.kafka.ZkCoordinator
>> - Task [1/1] New partition managers:
>> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>> partition=0},
>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>> partition=1},
>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>> partition=2}]
>> 42094 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - Read partition information from:
>> /brokers/rawWarehousePriceSpout/partition_0 --> null
>> 42363 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - No partition information found, using
>> configuration to determine offset
>> 42363 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - Starting Kafka
>> price-engine-demo-server.c.celertech-01.internal:0 from offset 425530
>> 42366 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - Read partition information from:
>> /brokers/rawWarehousePriceSpout/partition_1 --> null
>> 42400 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - No partition information found, using
>> configuration to determine offset
>> 42400 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - Starting Kafka
>> price-engine-demo-server.c.celertech-01.internal:1 from offset 550715
>> 42401 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - Read partition information from:
>> /brokers/rawWarehousePriceSpout/partition_2 --> null
>> 42432 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - No partition information found, using
>> configuration to determine offset
>> 42432 [Thread-15-rawWarehousePriceSpout] INFO
>> storm.kafka.PartitionManager - Starting Kafka
>> price-engine-demo-server.c.celertech-01.internal:2 from offset 439547
>> 42432 [Thread-15-rawWarehousePriceSpout] INFO storm.kafka.ZkCoordinator
>> - Task [1/1] Finished refreshing
>> 44652 [ProcessThread(sid:0 cport:-1):] INFO
>> org.apache.storm.zookeeper.server.PrepRequestProcessor - Got user-level
>> KeeperException when processing sessionid:0x14d76d4418d000c type:create
>> cxid:0x5 zxid:0x2c txntype:-1 reqpath:n/a Error
>> Path:/brokers/rawWarehousePriceSpout Error:KeeperErrorCode = NoNode for
>> /brokers/rawWarehousePriceSpout
>>
>>
>>
>> On 21 May 2015, at 15:02, Cristian Makoto Sandiga <cmsand...@gmail.com>
>> wrote:
>>
>> Im sorry, i made a mistake
>>
>> ZkHosts zkhost = new ZkHosts("localhost:2181","/brokers"); // /brokers
>> -> kafka broker
>> SpoutConfig spoutconfig = new SpoutConfig(zkhost, "bid_history",
>> "/brokers", "kafka-spout"); --> storm --> zookeeper
>>
>> If you see not is a ERROR, is a INFO usually when directory structure not
>> exist.
>>
>> Take a look Zookeeper Kaka-broker (port 2181)
>> ----------------------------------
>>
>> -- bin/zookeeper-shell.sh localhost:2181
>>
>> ls /brokers/topics/bid_history/partitions/0/state
>> []
>>
>> get /brokers/topics/bid_history/partitions/0/state
>> {"controller_epoch":1,"leader":0,"version":1,"leader_epoch":0,"isr":[0]}
>> cZxid = 0x113
>> ctime = Tue May 19 18:19:24 BRT 2015
>> mZxid = 0x113
>> mtime = Tue May 19 18:19:24 BRT 2015
>> pZxid = 0x113
>> cversion = 0
>> dataVersion = 0
>> aclVersion = 0
>> ephemeralOwner = 0x0
>> dataLength = 72
>> numChildren = 0
>>
>>
>> Take a look Zookeeper Storm-broker (port 2000)
>> ----------------------------------
>> Start topology, send a message and then look:
>>
>> bin/zookeeper-shell.sh localhost:2000
>> Connecting to localhost:2000
>> Welcome to ZooKeeper!
>> JLine support is disabled
>>
>> ls /
>> [storm, brokers, zookeeper]
>> ls /brokers
>> [kafka-spout]
>> ls /brokers/kafka-spout
>> [partition_0]
>> ls /brokers/kafka-spout/partition_0
>> []
>> get /brokers/kafka-spout/partition_0
>>
>> {"topology":{"id":"a9be1962-6b4e-4ed4-ae68-155a1948a1f6","name":"consolidate_reports"},"offset":4426029,"partition":0,"broker":{"host":"localhost","port":9092},"topic":"bid_history"}
>> cZxid = 0x50
>> ctime = Thu May 21 11:00:48 BRT 2015
>> mZxid = 0x50
>> mtime = Thu May 21 11:00:48 BRT 2015
>> pZxid = 0x50
>> cversion = 0
>> dataVersion = 0
>> aclVersion = 0
>> ephemeralOwner = 0x0
>> dataLength = 182
>> numChildren = 0
>>
>>
>> 2015-05-21 10:43 GMT-03:00 Benjamin Cuthbert <cuthbert....@gmail.com>:
>>
>>> Bec I do see this error in the storm logs
>>>
>>> 5304 [ProcessThread(sid:0 cport:-1):] INFO
>>> org.apache.storm.zookeeper.server.PrepRequestProcessor - Got user-level
>>> KeeperException when processing sessionid:0x14d76b1ea54000c type:create
>>> cxid:0x5 zxid:0x2c txntype:-1 reqpath:n/a Error
>>> Path:/brokers/rawWarehousePriceSpout Error:KeeperErrorCode = NoNode for
>>> /brokers/rawWarehousePriceSpout
>>>
>>>
>>>
>>> On 21 May 2015, at 14:04, Cristian Makoto Sandiga <cmsand...@gmail.com>
>>> wrote:
>>>
>>> Are you wrong
>>>
>>> zookeeperPath , // the root path in Zookeeper for the spout to store
>>> the consumer offsets
>>>
>>> Is for you KafkaSpout find information about broker meta information (
>>> topics, partitions)
>>>
>>> You have to override
>>>
>>> SpoutConfig spoutconfig = new SpoutConfig(zkhost, "bid_history",
>>> "/brokers", "kafka-spout");
>>> spoutconfig.zkServers
>>> spoutconfig.zkPort
>>>
>>> Take a look a KafkaSpout.class and do a debug mode in this part.
>>>
>>> 73 Map stateConf = new HashMap(conf);
>>> 74 List<String> zkServers = _spoutConfig.zkServers;
>>> if (zkServers == null) {
>>> zkServers = (List<String>)
>>> conf.get(Config.STORM_ZOOKEEPER_SERVERS);
>>> }
>>> Integer zkPort = _spoutConfig.zkPort;
>>> if (zkPort == null) {
>>> zkPort = ((Number)
>>> conf.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
>>> }
>>>
>>>
>>>
>>>
>>> 2015-05-21 9:53 GMT-03:00 Benjamin Cuthbert <cuthbert....@gmail.com>:
>>>
>>>> So when I run it out of eclipse in local mode I am pointing to the same
>>>> zookeeper as I am when it is deployed. Whatever way it still does not write
>>>> the offset.
>>>>
>>>> Only different I have is my spout is configured like so:
>>>>
>>>> BrokerHosts hosts = new ZkHosts(zookeeperQuorum);
>>>>
>>>> String zookeeperPath = KAFKA_STORM_DIR + "/" + topic;
>>>> SpoutConfig spoutConfig = new SpoutConfig(
>>>> hosts,
>>>> topic, // topic to read from
>>>> zookeeperPath , // the root path in Zookeeper for the spout to store
>>>> the consumer offsets
>>>> spoutId); // an id for this consumer for storing the consumer offsets
>>>> in Zookeeper
>>>>
>>>> //Check if we should be consuming messages from the beginning
>>>> spoutConfig.forceFromStart = consumeFromBeginning;
>>>> spoutConfig.maxOffsetBehind = Long.MAX_VALUE;
>>>> spoutConfig.useStartOffsetTimeIfOffsetOutOfRange = true;
>>>> spoutConfig.scheme = new SchemeAsMultiScheme(new
>>>> KafkaAnalyticsMessageDecoder());
>>>>
>>>>
>>>> On 21 May 2015, at 13:50, Cristian Makoto Sandiga <cmsand...@gmail.com>
>>>> wrote:
>>>>
>>>> I have the same problem when i have Storm in Local cluster mode. but is
>>>> because, Local mode has a zookeeper embedded and every time that restart
>>>> the offset is null and is the reason that you have
>>>> /kafkastorm/warehouse_prices/rawWarehousePriceSpout/partition_2 -->
>>>> null.
>>>>
>>>> Try to run Local Mode, send some messages to spout, and then see Offset
>>>> in Zookeeper (port 2000).
>>>>
>>>> Or use Zookeeper of kafka, override.
>>>>
>>>> spoutconfig.zkServers
>>>> spoutconfig.zkPort
>>>>
>>>>
>>>> 2015-05-21 9:39 GMT-03:00 Benjamin Cuthbert <cuthbert....@gmail.com>:
>>>>
>>>>> Thanks,
>>>>>
>>>>> Yup kafka is creating them on startup when the topics gets its first
>>>>> msg. And from below the storm logs are never really committing the offset
>>>>> to zookeeper.
>>>>>
>>>>> Zookeeper kafka topics details
>>>>>
>>>>> [zk: 127.0.0.1:2181(CONNECTED) 41] ls
>>>>> /brokers/topics/warehouse_prices/partitions
>>>>> [2, 1, 0]
>>>>> [zk: 127.0.0.1:2181(CONNECTED) 42]
>>>>>
>>>>> Zookeeper Storm
>>>>>
>>>>> [zk: 127.0.0.1:2181(CONNECTED) 42] ls
>>>>> /kafkastorm/warehouse_prices/rawWarehousePriceSpout
>>>>> []
>>>>> [zk: 127.0.0.1:2181(CONNECTED) 43]
>>>>>
>>>>>
>>>>> Storm logs
>>>>>
>>>>> 44325 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] New partition managers:
>>>>> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=0},
>>>>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=1},
>>>>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=2}]
>>>>> 44491 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Read partition information from:
>>>>> /kafkastorm/warehouse_prices/rawWarehousePriceSpout/partition_0 --> null
>>>>> 44746 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - No partition information found, using
>>>>> configuration to determine offset
>>>>> 44746 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Last commit offset from zookeeper: 0
>>>>> 44747 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Commit offset 0 is more than
>>>>> 9223372036854775807 behind, resetting to startOffsetTime=-2
>>>>> 44747 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Starting Kafka
>>>>> price-engine-demo-server.c.celertech-01.internal:0 from offset 0
>>>>> 44749 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Read partition information from:
>>>>> /kafkastorm/warehouse_prices/rawWarehousePriceSpout/partition_1 --> null
>>>>> 44778 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - No partition information found, using
>>>>> configuration to determine offset
>>>>> 44778 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Last commit offset from zookeeper: 0
>>>>> 44778 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Commit offset 0 is more than
>>>>> 9223372036854775807 behind, resetting to startOffsetTime=-2
>>>>> 44779 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Starting Kafka
>>>>> price-engine-demo-server.c.celertech-01.internal:1 from offset 0
>>>>> 44781 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Read partition information from:
>>>>> /kafkastorm/warehouse_prices/rawWarehousePriceSpout/partition_2 --> null
>>>>> 44809 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - No partition information found, using
>>>>> configuration to determine offset
>>>>> 44809 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Last commit offset from zookeeper: 0
>>>>> 44810 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Commit offset 0 is more than
>>>>> 9223372036854775807 behind, resetting to startOffsetTime=-2
>>>>> 44810 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.PartitionManager - Starting Kafka
>>>>> price-engine-demo-server.c.celertech-01.internal:2 from offset 0
>>>>> 44810 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Finished refreshing
>>>>> 47438 [ProcessThread(sid:0 cport:-1):] INFO
>>>>> org.apache.storm.zookeeper.server.PrepRequestProcessor - Got user-level
>>>>> KeeperException when processing sessionid:0x14d7658db3c000c
>>>>> type:create cxid:0x5 zxid:0x2c txntype:-1 reqpath:n/a Error
>>>>> Path:/kafkastorm/warehouse_prices/rawWarehousePriceSpout
>>>>> Error:KeeperErrorCode = NoNode for
>>>>> /kafkastorm/warehouse_prices/rawWarehousePriceSpout
>>>>> 104184 [Thread-15-rawWarehousePriceSpout] WARN storm.kafka.KafkaUtils
>>>>> - No data found in Kafka Partition partition_0
>>>>> 104828 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Refreshing partition manager
>>>>> connections
>>>>> 105005 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.DynamicBrokersReader - Read partition info from zookeeper:
>>>>> GlobalPartitionInformation{partitionMap={0=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> 1=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> 2=price-engine-demo-server.c.celertech-01.internal:9092}}
>>>>> 105005 [Thread-15-rawWarehousePriceSpout] INFO storm.kafka.KafkaUtils
>>>>> - Task [1/1] assigned
>>>>> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=0},
>>>>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=1},
>>>>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=2}]
>>>>> 105006 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Deleted partition managers: []
>>>>> 105006 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] New partition managers: []
>>>>> 105006 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Finished refreshing
>>>>> 164204 [Thread-15-rawWarehousePriceSpout] WARN storm.kafka.KafkaUtils
>>>>> - No data found in Kafka Partition partition_0
>>>>> 165063 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Refreshing partition manager
>>>>> connections
>>>>> 165240 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.DynamicBrokersReader - Read partition info from zookeeper:
>>>>> GlobalPartitionInformation{partitionMap={0=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> 1=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> 2=price-engine-demo-server.c.celertech-01.internal:9092}}
>>>>> 165240 [Thread-15-rawWarehousePriceSpout] INFO storm.kafka.KafkaUtils
>>>>> - Task [1/1] assigned
>>>>> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=0},
>>>>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=1},
>>>>> Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
>>>>> partition=2}]
>>>>> 165240 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Deleted partition managers: []
>>>>> 165240 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] New partition managers: []
>>>>> 165240 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Finished refreshing
>>>>> 224233 [Thread-15-rawWarehousePriceSpout] WARN storm.kafka.KafkaUtils
>>>>> - No data found in Kafka Partition partition_0
>>>>> 225308 [Thread-15-rawWarehousePriceSpout] INFO
>>>>> storm.kafka.ZkCoordinator - Task [1/1] Refreshing partition manager
>>>>> connections
>>>>>
>>>>>
>>>>> On 21 May 2015, at 13:28, Cristian Makoto Sandiga <cmsand...@gmail.com>
>>>>> wrote:
>>>>>
>>>>> Zookeeper create nothing when startup, you have to create your
>>>>> partitions in kafka broker.
>>>>>
>>>>> bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic
>>>>> click_history --replication-factor 1 --partitions 10
>>>>>
>>>>>
>>>>>
>>>>> 2015-05-21 8:58 GMT-03:00 Benjamin Cuthbert <cuthbert....@gmail.com>:
>>>>>
>>>>>> All,
>>>>>>
>>>>>> We changed or paths in zookeeper and we are now seeing
>>>>>>
>>>>>> java.lang.RuntimeException: java.lang.RuntimeException:
>>>>>> org.apache.zookeeper.KeeperException$NoNodeException:
>>>>>> KeeperErrorCode = NoNode for /brokers/topics/warehouse_prices/partitions
>>>>>> at storm.kafka.DynamicBrokersReader.getBrokerInfo(
>>>>>> DynamicBrokersReader.java:81) ~[storm-kafka-0.9.4.jar:0.9.4]
>>>>>> at storm.kafka.trident.ZkBrokerReader.<init>(ZkBrokerReader.java:42)
>>>>>> ~[storm-kafka-0.9.4.jar:0.9.4]
>>>>>> at storm.kafka.KafkaUtils.makeBrokerReader(KafkaUtils.java:57)
>>>>>> ~[storm-kafka-0.9.4.jar:0.9.4]
>>>>>> at storm.kafka.KafkaSpout.open(KafkaSpout.java:87)
>>>>>> ~[storm-kafka-0.9.4.jar:0.9.4]
>>>>>> at
>>>>>> backtype.storm.daemon.executor$fn__3371$fn__3386.invoke(executor.clj:522)
>>>>>> ~[storm-core-0.9.4.jar:0.9.4]
>>>>>> at backtype.storm.util$async_loop$fn__460.invoke(util.clj:461)
>>>>>> ~[storm-core-0.9.4.jar:0.9.4]
>>>>>> at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
>>>>>> at java.lang.Thread.run(Thread.java:744) [na:1.7.0_51]
>>>>>>
>>>>>> Should this not be discovered by the Spout on startup?
>>>>>>
>>>>>
>>>>>
>>>>>
>>>>
>>>>
>>>
>>>
>>
>>
>
>
