Thanks or the replies Jason! A couple of follow up questions: i. If I call seekToEnd() on the consumer and then use position() is the position() call making a blocking IO call. It is not entirely clear from the documentation whether this will just block or not.
ii. If the position() call does indeed block are there any consequences in terms of me not calling poll often enough? Is there a way for me to limit how long I allow it to block, maybe by passing a timeout parameter. I only use manual assignments so I am hoping that there is no consequence of infrequent heart beats etc through poll starvation. Thanks, Rajiv On Wed, Jan 6, 2016 at 1:58 PM, Jason Gustafson <ja...@confluent.io> wrote: > Hi Rajiv, > > Answers below: > > i) How do I get the last log offset from the Kafka consumer? > > > To get the last offset, first call seekToEnd() and then use position(). > > ii) If I ask the consumer to seek to the beginning via the consumer > > .seekToBeginning(newTopicPartition) call, will it handle the case where > > the > > log has rolled over in the meanwhile and what was considered the > beginning > > offset is no longer present? > > > The call to seekToBeginning() only sets a flag indicating that a reset is > needed. The actual position will not be fetched until you call poll() or > position(). This means that the window for an out of range offset should be > small, but of course it could happen. The behavior of the consumer when an > offset is out of range is controlled with the "auto.offset.reset" > configuration. If you use the "earliest" policy, then the consumer will > automatically reset the position to whatever the current earliest offset > is. You might also choose to use no automatic reset policy by specifying > "none." In this case, poll() will throw an OffsetOutOfRangeException, which > you can catch and manually re-seek to the beginning. > > iii) What settings do I need on the Kafka broker (besides > > log.retention.minutes = 10) to ensure that my partitions don't retain any > > more than 10 minutes of data (plus a couple minutes is fine). Do I need > to > > tune how often Kafka checks for log deletion eligibility? Any other > > settings I should play with to ensure timely deletion? > > > Log retention is currently at the granularity of log segments. This means > that you cannot generally guarantee that messages will be deleted within > the configured retention time. However, you can control the segment size > using "log.segment.bytes" and the delay before deletion with " > log.segment.delete.delay.ms." If you can estimate the incoming message > rate, then you can probably tune these settings to get a retention policy > closer to what you're looking for. See here for more info on broker > configuration: https://kafka.apache.org/documentation.html#brokerconfigs. > And for what it's worth, KIP-32, which adds a timestamp to each message, > should provide some better options for handling this. > > -Jason > > On Wed, Jan 6, 2016 at 9:37 AM, Rajiv Kurian <ra...@signalfx.com> wrote: > > > I want to use the new 0.9 consumer for a particular application. > > > > My use case is the following: > > > > i) The TopicPartition I need to poll has a short log say 10 mins odd > > (log.retention.minutes is set to 10). > > > > ii) I don't use a consumer group i.e. I manage the partition assignment > > myself. > > > > iii) Whenever I get a new partition assigned to one of my processes > > (through my own mechanism), I want to query for the current end of the > log > > and then seek to the beginning of the log. I want to continue reading in > a > > straight line till my offset moves from the beginning to the end that I > > queried before beginning to poll. When I am done reading this much data > (I > > know the end has moved by the time I've read all of it) I consider myself > > caught up. Note: I only need to do the seek to the beginning of the log, > > which the new consumer allows one to do. I just need to know the end of > log > > offset so that I know that I have "caught up". > > > > So questions I have are: > > > > i) How do I get the last log offset from the Kafka consumer? The > > SimpleConsumer had a way to get this information. If I can get this info > > from the consumer, I plan to do something like this: > > > > > > private boolean assignNewPartitionAndCatchUp(int newPartition) { > > > > final TopicPartition newTopicPartition = new > > TopicPartition(myTopic, > > newPartition); > > > > // Queries the existing partitions and adds this TopicPartition to > > the list. > > > > List<TopicPartition> newAssignment = > > createNewAssignmentByAddingPartition( > > > > newTopicPartition); > > > > consumer.assign(newAssignment); > > > > > > // How do I actually do this with the consumer? > > > > final long lastMessageOffset = getLastMessageOffset( > > newTopicPartition); > > > > > > consumer.seekToBeginning(newTopicPartition); > > > > final long timeout = 100; > > > > int numIterations = 0; > > > > final boolean caughtUp = false; > > > > while (!caughtUp && numIterations < maxIterations) { > > > > ConsumerRecords<Void, byte[]> records = > consumer.poll(timeout); > > > > numIterations += 1; > > > > for (ConsumerRecord<Void, byte[]> record : records) { > > > > // All messages are processed regularly even if they > belong > > to other partitions. > > > > processRecord(record.value()); > > > > final int partition = record.partition(); > > > > final long offset = record.offset(); > > > > // Only if we find that the new partition has caught up > do > > we return. > > > > if (partition == newPartition && offset >= > > lastMessageOffset) > > { > > > > caughtUp = true; > > > > } > > > > } > > > > } > > > > return caughtUp; > > > > } > > > > ii) If I ask the consumer to seek to the beginning via the consumer > > .seekToBeginning(newTopicPartition) call, will it handle the case where > the > > log has rolled over in the meanwhile and what was considered the > beginning > > offset is no longer present. Given my log retention is only 10 minutes > and > > the partitions will each get quite a bit of traffic, I'd imagine that > > messages will fall out of the log quite often. > > > > iii) What settings do I need on the Kafka broker (besides > > log.retention.minutes = 10) to ensure that my partitions don't retain any > > more than 10 minutes of data (plus a couple minutes is fine). Do I need > to > > tune how often Kafka checks for log deletion eligibility? Any other > > settings I should play with to ensure timely deletion? > > > > Thanks, > > Rajiv > > >
