On Thu, Jan 7, 2016 at 12:54 PM, Heath Ivie <hi...@autoanything.com> wrote:
> 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?
>
You can run multiple consumer instances in one or more consumer groups on a
single REST proxy instance. However, you'll need to ensure they have
different IDs (which you can have auto-assigned if you like).
>
> 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.
>
>
How many partitions does your topic have? Did you create it before this
test or was it auto-created? One reason you could see this behavior is if
you only have 1 partition. In that case only 1 consumer would see any
messages since only 1 consumer would have any partitions assigned to it.
-Ewen
> 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.
>
--
Thanks,
Ewen
