[
https://issues.apache.org/jira/browse/KAFKA-546?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13490977#comment-13490977
]
Jun Rao commented on KAFKA-546:
-------------------------------
Thanks for patch v2. Some comments:
20. ConsumerIterator.next(): The following code depends on no gaps in offsets.
This is true at this moment, but may not be true in the future when we have a
different retention policy. A safer way is to keep iterating the messageSet
until we get an offset that reaches or passes ctiConsumeOffset.
for (i <- 0L until (ctiConsumeOffset - cdcFetchOffset)) {
localCurrent.next()
21. PartitionTopicInfo: In startOffset(), unfortunately, we can't use the
shallow iterator. This is because when messages are compressed, the offset of
the top level message has the offset of the last message (instead of the first
one) in the compressed unit. Also, iterating messages here may not be ideal
since it forces us to decompress. An alternative way is to do the logic in
ConsumerIterator.next(). Everytime that we get a new chunk of messageset, we
keep iterating it until the message offset reaches or passes the
consumeroffset. This way, if we are doing shallow iteration, we don't have to
decompress messages.
22. ConsumerIteratorTest:
22.1 zkConsumerConnector is not used.
22.2 We probably should set consumerOffset to a value >0 in PartitionTopicInfo.
22.3 Also, could we add a test that covers compressed messageset?
> Fix commit() in zk consumer for compressed messages
> ---------------------------------------------------
>
> Key: KAFKA-546
> URL: https://issues.apache.org/jira/browse/KAFKA-546
> Project: Kafka
> Issue Type: New Feature
> Affects Versions: 0.8
> Reporter: Jay Kreps
> Assignee: Swapnil Ghike
> Attachments: kafka-546-v1.patch, kafka-546-v2.patch
>
>
> In 0.7.x and earlier versions offsets were assigned by the byte location in
> the file. Because it wasn't possible to directly decompress from the middle
> of a compressed block, messages inside a compressed message set effectively
> had no offset. As a result the offset given to the consumer was always the
> offset of the wrapper message set.
> In 0.8 after the logical offsets patch messages in a compressed set do have
> offsets. However the server still needs to fetch from the beginning of the
> compressed messageset (otherwise it can't be decompressed). As a result a
> commit() which occurs in the middle of a message set will still result in
> some duplicates.
> This can be fixed in the ConsumerIterator by discarding messages smaller than
> the fetch offset rather than giving them to the consumer. This will make
> commit work correctly in the presence of compressed messages (finally).
--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira
