Thanks Jason and Craig. Jason, that is exactly correct. I am using the simple consumer mode. Is there an easy way to find out the lag for consumers using the Simple Consumer mode?
Regards, Ryan On Fri, Mar 25, 2016 at 11:47 AM, Jason Gustafson <[email protected]> wrote: > Hey Ryan, > > Sounds like you might be using the so-called "simple consumer" mode. If you > use assign() to give your consumer a specific partition, you're not > actually using a consumer group, so there won't be any coordination with > other consumers. If you use subscribe() on the other hand, then you should > see the group listed in the output of consumer-groups.sh. > > -Jason > > On Thu, Mar 24, 2016 at 7:37 PM, craig w <[email protected]> wrote: > >> Did you subscribe and poll? I believe your consumer group won't show up >> until it has been assigned one or more partitions. >> On Mar 24, 2016 9:48 PM, "Ryan Phillips" <[email protected]> wrote: >> >> > I am only assigning this consumer one partition to listen to. Perhaps >> > that is where the issue lies, because the kafka-console-consumer's >> > group id (listening on all partitions) shows up correctly within the >> > kafka-consumer-groups command. >> > >> > On Thu, Mar 24, 2016 at 6:59 PM, Ryan Phillips <[email protected]> >> wrote: >> > > Greetings, >> > > >> > > I’m attempting to use the New Consumer in my project (0.9.0.1), and it >> > > appears to be working nicely. The issue that I am seeing is that the >> > > group.id is not showing up within: >> > > >> > > kafka-consumer-groups.sh —bootstrap-server >> > > kafka0.local:9092,kafka1.local:9092 —new-consumer —list >> > > >> > > The command loads up, but the consumer that is running is not >> > > displayed. The consumer is clearly processing a massive amount of >> > > messages. I have included the properties used at the end of my >> > > message. >> > > >> > > Thanks, >> > > Ryan >> > > >> > > Properties props = new Properties(); >> > > props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); >> > > props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBrokerSeeds); >> > > props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); >> > > props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); >> > > props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 6 * 1024 * >> > 1024); >> > > props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30 * 1000); >> > > props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15 * 1000); >> > > props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, >> > > "org.apache.kafka.common.serialization.StringDeserializer"); >> > > props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, >> > > "org.apache.kafka.common.serialization.ByteArrayDeserializer"); >> > > consumer = new KafkaConsumer<String, byte[]>(props); >> > >>
