Hi Adrian,
KafkaSpout is a consumer in this case you would be
connecting to zookeeper and KafkaBolt which is a
producer(kafka) needs to connect to a list of broker
(localhost:9092). KafkaSpout uses SpoutConfig in which you can
add to List<String>zkServers and for KafkaBolt you can create a
Properties object with "metadata.broker.list" and pass a comma
separated strings.
-Harsha
On Tue, Jul 29, 2014, at 05:11 AM, Adrian Landman wrote:
I feel like you missed the issue in my question. For the
connection string for ZkHosts, if I pass in localhost:9092 with
a default kafka configuration, it won't connect. Instead I
have to pass in localhost:2181. Is this expected behavior?
Also, if I wanted to pass in more than one host, what should
separate the entries? Commas?
On Mon, Jul 28, 2014 at 4:17 PM, Parth Brahmbhatt
<[1][email protected]> wrote:
For setting a list of brokers in kafkaSpout, I believe there
are 2 options:
If you use StaticHosts then you need to add
GlobalPartitionInformation in which you have to specify each
partition and its corresponding broker host
GlobalPartitionInformation partitions = new
GlobalPartitionInformation();
partitions.addPartition(0,new Broker("10.22.2.79", 9092));
//add more partitions here.
BrokerHosts hosts = new StaticHosts(partitions);
Alternatively, If you use ZkHosts then you need to pass in the
complete zookeeper connection string e.g. lolcalhost:9092 ,
optionally there is a constructor that allows you to specify a
second argument which is zkroot , by default it is set
to /brokers which should work with default kafka installation.
The code in ZlHosts looks under zkroot/topics/<topic that you
set in SpoutConfig>/partitions to figure out the number of
partitions and leader for each partition.
Thanks
Parth
On Mon, Jul 28, 2014 at 9:54 AM, Adrian Landman
<[2][email protected]> wrote:
I am writing a topology that pulls messages from a topic, does
some work, and then writes them back on a different topic. I
have been having some issues so I created my own small topology
that just pulls a message, prints the contents, and then stores
them back on a new topic. I finally got this to work, but it
raised a question.
To create the spout I need to either pass in the kafka location
sans port (e.g. localhost) or use 2181 as the port.
To create the producer bolt I need to pass in the broker port
(e.g. 9092) or I get an array out of bounds exception when
creating the Producer.
When we were using kafka7 and
[3]https://github.com/nathanmarz/storm-contrib/tree/master/stor
m-kafka/src/jvm/storm/kafka
for our storm/kafka integration I believe that we used the same
broker list for both our spout and our producer. Is there
anyway to do the same with kafka8 and the new storm-kafka
project? Also, if w want to pass in a list, I know that in the
KafkaBolt we can set metadata.broker.list to a comma separated
list of brokers ([4]1.1.1.1:9092, [5]1.1.1.2:9092) but can we
do the same for the spout? Or is there any reason to? ZkHost
takes in a String, but I didn't see anything that specified the
format.
--
Thanks
Parth
CONFIDENTIALITY NOTICE
NOTICE: This message is intended for the use of the individual
or entity to which it is addressed and may contain information
that is confidential, privileged and exempt from disclosure
under applicable law. If the reader of this message is not the
intended recipient, you are hereby notified that any printing,
copying, dissemination, distribution, disclosure or forwarding
of this communication is strictly prohibited. If you have
received this communication in error, please contact the sender
immediately and delete it from your system. Thank You.
References
1. mailto:[email protected]
2. mailto:[email protected]
3.
https://github.com/nathanmarz/storm-contrib/tree/master/storm-kafka/src/jvm/storm/kafka
4. http://1.1.1.1:9092/
5. http://1.1.1.2:9092/
