the map in the message manager constructor is from Storm on prepare
On Wed, Feb 24, 2016 at 6:34 PM, Brian Taylor <
[email protected]> wrote:
> and partitioner:
>
> package com.company.rt.producers.utilities.partitioners;
>
> import java.util.concurrent.atomic.AtomicInteger;
>
> import kafka.producer.Partitioner;
> import kafka.utils.VerifiableProperties;
>
> import org.slf4j.Logger;
> import org.slf4j.LoggerFactory;
>
> /**
> *
> */
> public class KafkaRoundRobinPartitioner implements Partitioner {
>
> private static final Logger LOG =
> LoggerFactory.getLogger(KafkaRoundRobinPartitioner.class);
>
> final AtomicInteger counter = new AtomicInteger(0);
>
> public KafkaRoundRobinPartitioner(VerifiableProperties props) {
> LOG.info("Instantiated the Storm Round Robin Partitioner class with
> properties: \n");
> for(String propName : props.props().stringPropertyNames()) {
> LOG.info("{}:{} \n", propName, props.props().getProperty(propName));
> }
> }
>
> @Override
> public int partition(Object key, int numberOfPartitions) {
> int partitionId = counter.incrementAndGet() % numberOfPartitions;
> if (counter.get() > Integer.MAX_VALUE) {
> counter.set(0);
> }
> return partitionId;
> }
> }
>
>
> On Wed, Feb 24, 2016 at 6:33 PM, Brian Taylor <
> [email protected]> wrote:
>
>> Here's an example Kafka client:
>>
>> package com.company.rt.services.data;
>>
>> import java.util.ArrayList;
>> import java.util.List;
>> import java.util.Map;
>> import java.util.Properties;
>>
>> import com.company.models.Event;
>> import kafka.javaapi.producer.Producer;
>> import kafka.producer.KeyedMessage;
>> import kafka.producer.ProducerConfig;
>>
>> /**
>> *
>> */
>> public final class MessageManager {
>>
>> private Properties relaxedProducerProps;
>> private ProducerConfig relaxedProducerConfig;
>> private Producer<String,String> relaxedProducer;
>>
>> private Properties durableProducerProps;
>> private ProducerConfig durableProducerConfig;
>> private Producer<String,String> durableProducer;
>>
>> public MessageManager(Map map) {
>> relaxedProducerProps = new Properties();
>> relaxedProducerProps.put("metadata.broker.list",
>> map.get("topology.kafka.broker.list"));
>> relaxedProducerProps.put("serializer.class",
>> "kafka.serializer.StringEncoder");
>> relaxedProducerProps.put("key.serializer.class",
>> "kafka.serializer.StringEncoder");
>> relaxedProducerProps.put("request.required.acks", "1");
>> relaxedProducerProps.put("partitioner.class",
>> "com.company.rt.producers.utilities.partitioners.KafkaRoundRobinPartitioner");
>> relaxedProducerConfig = new ProducerConfig(relaxedProducerProps);
>> relaxedProducer = new Producer<>(relaxedProducerConfig);
>>
>> durableProducerProps = new Properties();
>> durableProducerProps.put("metadata.broker.list",
>> map.get("topology.kafka.broker.list"));
>> durableProducerProps.put("serializer.class",
>> "kafka.serializer.StringEncoder");
>> durableProducerProps.put("key.serializer.class",
>> "kafka.serializer.StringEncoder");
>> durableProducerProps.put("request.required.acks", "-1");
>> durableProducerProps.put("partitioner.class",
>> "com.company.rt.producers.utilities.partitioners.KafkaRoundRobinPartitioner");
>> durableProducerConfig = new ProducerConfig(durableProducerProps);
>> durableProducer = new Producer<>(durableProducerConfig);
>> }
>>
>> public void send(String topic, String message, boolean durable) throws
>> Exception {
>> KeyedMessage<String,String> keyedMessage = new
>> KeyedMessage<>(topic,"key",message);
>> if (durable) durableProducer.send(keyedMessage);
>> else relaxedProducer.send(keyedMessage);
>> }
>>
>> public void send(String topic, List<String> messages, boolean durable)
>> throws Exception {
>> List<KeyedMessage<String, String>> keyedMessages = new ArrayList<>();
>> for(String message : messages) {
>> keyedMessages.add(new KeyedMessage<>(topic,"key",message));
>> }
>> if (durable) durableProducer.send(keyedMessages);
>> else relaxedProducer.send(keyedMessages);
>> }
>>
>> public void send(String topic, Event event, boolean durable) throws
>> Exception {
>> // KeyedMessage<String, Event> keyedMessage = new
>> KeyedMessage<>(topic, "key", event);
>> // if(durable) durableProducer.send(keyedMessage);
>> // else relaxedProducer.send(keyedMessage);
>> }
>>
>> }
>>
>>
>> On Wed, Feb 24, 2016 at 3:00 PM, david kavanagh <[email protected]>
>> wrote:
>>
>>> Thanks for the reply guys.
>>>
>>> I have looked at kafka-node and it looks reasonable simple to integrate
>>> Node with Storm. My original problem was getting the KafkaBolt to work, so
>>> i can output the data to Kafka and then use kafka-node to consume the data
>>> to the Node app. I have had some help from this mailing list trying to get
>>> the KafkaBolt to work but there seems to be a java issue (nothing to with
>>> Storm) that i can't get to the bottom of. I am doing this as part of my
>>> final year project in college and i am brand new to all these technologies
>>> (apart from Java) so it is more than likely something i am doing wrong
>>> somewhere.
>>>
>>> I don't actually need to use Kafka. That just seemed like the best way
>>> to connect Storm to the app. I have just taken a quick look at Redis and it
>>> seems to be a better way of connecting the two. I will try to get that up
>>> and running tomorrow and i will post up how i get on. Thanks again, the
>>> advice is much appreciated.
>>>
>>> Regards,
>>> David
>>>
>>> ------------------------------
>>> Date: Wed, 24 Feb 2016 14:31:40 -0500
>>> Subject: Re: Connecting Storm Output to Node.js app
>>> From: [email protected]
>>> To: [email protected]
>>>
>>>
>>> Have you tried https://www.npmjs.com/package/kafka-node
>>>
>>> On Wed, Feb 24, 2016 at 2:28 PM, Patrick Wiener <[email protected]>
>>> wrote:
>>>
>>> Have you considered using Redis?
>>>
>>> You could hook up Storm with Redis (e.g. through Jedis) and using Redis
>>> built in pub/sub mechanism for data retrieval from within Node.js
>>>
>>>
>>> Regards,
>>> Patrick
>>>
>>>
>>> Am 23.02.2016 um 19:47 schrieb david kavanagh <[email protected]>:
>>>
>>> Hello,
>>>
>>> I currently have Storm up and running on two Ubuntu VM's. I have Storm
>>> pulling data from a database on a MySQL Cluster and now I am trying to
>>> figure out a way of getting Storm and a Node.js app on a remote VM
>>> communicating. I have tried to use a Kafka Topic to store the output
>>> from Storm so the app can connect to the topic and gather the data,
>>> but i cannot get the KafkaBolt to work as it should. I have tried
>>> everything
>>> i can think of but no luck. Is there another way of getting Storm to
>>> connect
>>> to a remote Node.js app? I have searched exhaustively online but cannot
>>> find
>>> anything. If there are some tutorials i could be pointed to that would
>>> be great.
>>> Any help at all would be greatly appreciated.
>>>
>>> Kind Regards
>>> David
>>>
>>>
>>>
>>>
>>>
>>> --
>>>
>>> Warm Regards,
>>> Brian Taylor
>>>
>>> Resolving Architecture .:.
>>>
>>> 330-812-7098
>>>
>>> [email protected]
>>>
>>> http://resolvingarchitecture.com
>>>
>>> www.linkedin.com/in/javadevops/
>>>
>>
>>
>>
>> --
>>
>> Warm Regards,
>> Brian Taylor
>>
>> Resolving Architecture .:.
>>
>> 330-812-7098
>>
>> [email protected]
>>
>> http://resolvingarchitecture.com
>>
>> www.linkedin.com/in/javadevops/
>>
>
>
>
> --
>
> Warm Regards,
> Brian Taylor
>
> Resolving Architecture .:.
>
> 330-812-7098
>
> [email protected]
>
> http://resolvingarchitecture.com
>
> www.linkedin.com/in/javadevops/
>
--
Warm Regards,
Brian Taylor
Resolving Architecture .:.
330-812-7098
[email protected]
http://resolvingarchitecture.com
www.linkedin.com/in/javadevops/
