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