Kafka-Rest question

[Heath Ivie]

Hi,

I am building my application using the rest proxy and I wanted to get some 
clarification on the documentation.

"Because consumers are stateful, any consumer instances created with the REST 
API are tied to a specific REST proxy instance. A full URL is provided when the 
instance is created and it should be used to construct any subsequent requests. 
Failing to use the returned URL for future consumer requests will end up adding 
new consumers to the group. If a REST proxy instance is shutdown, it will 
attempt to cleanly destroy any consumers before it is terminated."

Does the highlighted mean that I have to have multiple deployed instances in 
order to have more than 1 consumer group and configure it with sticky sessions?

I have the following code where I am testing the behavior when I have 2 
consumers in the same group, and I am getting an empty array back for the 
second consumer in the same group.

This will through an exception when I attempt to read a message from the 2nd 
consumer.

Not sure if it would help, but I configured the proxy to have 2 threads per 
consumer.

KafkaMessageQueueWriter writer = new 
KafkaMessageQueueWriter("http://10.1.30.48:8082";);



            KafkaMessageQueueReader reader1 = new KafkaMessageQueueReader(new 
KafkaMessageQueueReaderSettings() { ConsumerGroupName = "Test50", TopicName = 
"ShellComparisonTest", ZookeeperClientPort = 8082, ZookeeperClientPortAddress = 
"10.1.30.48" });
            KafkaMessageQueueReader reader2 = new KafkaMessageQueueReader(new 
KafkaMessageQueueReaderSettings() { ConsumerGroupName = "Test51", TopicName = 
"ShellComparisonTest", ZookeeperClientPort = 8082, ZookeeperClientPortAddress = 
"10.1.30.48" });
            KafkaMessageQueueReader reader3 = new KafkaMessageQueueReader(new 
KafkaMessageQueueReaderSettings() { ConsumerGroupName = "Test50", TopicName = 
"ShellComparisonTest", ZookeeperClientPort = 8082, ZookeeperClientPortAddress = 
"10.1.30.48" });



            for (int i = 0; i < 10; i++)
            {
                //Publish a message to the ShellComparisonTest topic
                writer.PublishMessage("ShellComparisonTest", new OrderEvent() { 
OrderId = 5, OrderItemId = 1234 });
                var @event = 
reader1.ReadQueue<OrderEvent>("ShellComparisonTest");
                if (!@event.Any()) throw new Exception();

                @event = reader2.ReadQueue<OrderEvent>("ShellComparisonTest");
                if (!@event.Any()) throw new Exception();

                //Publish a message to the ShellComparisonTest topic so that
                //there is one unread message in the queue before trying to
                //read from the second consumer in the group
                writer.PublishMessage("ShellComparisonTest", new OrderEvent() { 
OrderId = 5, OrderItemId = 1234 });
                @event = reader3.ReadQueue<OrderEvent>("ShellComparisonTest");
                if (!@event.Any()) throw new Exception();
            }

Is there something that I am missing or is there a good resource to read.

Thanks

Heath


Warning: This e-mail may contain information proprietary to AutoAnything Inc. 
and is intended only for the use of the intended recipient(s). If the reader of 
this message is not the intended recipient(s), you have received this message 
in error and any review, dissemination, distribution or copying of this message 
is strictly prohibited. If you have received this message in error, please 
notify the sender immediately and delete all copies.