Hi Andrew, Thanks for the nice KIP on a very interesting feature about introducing some of the traditional MessageQueue semantics to Kafka. It is good to see that we are extending the existing consumer groups concepts and related mechanisms for shared subscriptions instead of bringing any large architectural/protocol changes.
This KIP talks about introducing a durable subscription feature for topics with multiple consumers consuming messages parallely from a single topic partition. 101 Are you planning to extend this functionality for queueing semantics like JMS point to point style in future? 102 When a message is rejected by the target consumer, how do users know what records/offsets are dropped because of the failed records due to rejection ack or due to timeouts etc before DLQs are introduced? 103 It talks about SPSO values, earliest being the default and user can reset it to a target offset timestamp. What is the maximum value for SPEO? It is good to clarify what could be the maximum value for SPSO and SPEO. It can be HW or LogStableOffset or some other value? 104 KIP mentions that "share.delivery.count.limit" as the maximum number of delivery attempts for a record delivered to a share group. But the actual delivery count may be more than this number as the leader may fail updating the delivery count as leader or consumer may fail and more delivery attempts may be made later. It may be the minimum number of delivery attempts instead of the maximum delivery attempts. Thanks, Satish. On Wed, 24 May 2023 at 21:26, Andrew Schofield <andrew_schofield_j...@outlook.com> wrote: > > Hi Stanislav, > Thanks for your email. You bring up some interesting points. > > 1) Tiered storage > I think the situation here for fetching historical data is equivalent to what > happens if a user resets the committed offset for a consumer > group back to an earlier point in time. So, I will mention this in the next > update to the KIP document but I think there's nothing > especially different here. > > 2) SSO initialized to the latest offset > The KIP does mention that it is possible for an administrator to set the SSO > using either AdminClient.alterShareGroupOffsets or > kafka-share-groups.sh. It is entirely intentional that there is no > KafkaConsumer config for initializing the SSO. I know that's how it > can be done for consumer groups, but it suffers from the situation where > different consumers have different opinions about > the initial value (earliest vs latest) and then the first one in wins. Also, > KIP-842 digs into some problems with how consumer > group offset reset works > (https://cwiki.apache.org/confluence/display/KAFKA/KIP-842%3A+Add+richer+group+offset+reset+mechanisms) > so > I've tried to sidestep those problems too. > > Another possibility is to follow KIP-848 which proposes that > AdminClient.incrementalAlterConfigs is enhanced to support a new > resource type called GROUP and supporting a dynamic group config in this > manner would give a single point of control. > > 3) Durable storage > The KIP does not yet describe how durable storage works. I have a few ideas > that I want to flesh out before updating the KIP. > > I will rule out using a compacted topic though. The problem is that each > record on a compacted topic is a key:value pair, and > it's not obvious what to use as the key. If it's the share group name, it > needs the entire in-flight record state to be recorded in > one hit which is extremely inefficient. > > 4) Batch acknowledgement > You are correct that compression makes delivery and acknowledgement of > individual messages within a compressed batch > more complicated. Again, I will defer a proper answer here until I've dug > more deeply. > > 5) Member management > Member management will be similar to consumer groups. I anticipate that it > will build entirely on the new consumer group > protocol in KIP-848. There seems little benefit in supporting the legacy > protocol when this KIP is targeting versions of Kafka > which will all have KIP-848. > > The two cases you mention: > i) If a bad consumer doesn't even heartbeat, it will be ejected from the > group. This does not involve a rebalance. > ii) If a bad consumer heartbeats but always times out message processing, it > will slow the advancement of the SSO/SEO. There > is the possibility that such a consumer would invalidate completely valid > messages. In order to do this, it would have to acquire > the same set of message repeatedly, to the exclusion of other consumers, and > thus bump the delivery count to the limit. > This is unlikely but not impossible. > > 6) Processing semantics > Delivery is at-least-once. > > 7) Acronyms > I hadn't thought about the implications of "Kafka SEO". I think I'll change > it to "Share Partition Start Offset" (SPSO) and > "Share Partition End Offset" (SPEO). > > There is a lot of work ahead for this KIP. I intend to work on the protocol > changes next. > > Thanks for getting involved in the discussion. > Andrew > > From: Stanislav Kozlovski <stanis...@confluent.io.INVALID> > Sent: 22 May 2023 11:20 > To: dev@kafka.apache.org <dev@kafka.apache.org> > Subject: Re: [DISCUSS] KIP-932: Queues for Kafka > > Hey Andrew! > > Kudos on the proposal. It is greatly written - a joy to read. It is > definitely an interesting solution to the queueing problem - I would not > have guessed we could solve it like this. Thank you for working on this. > > Happy to get the discussion started - I have a few comments/questions on > first read: > > 1. Tiered Storage > > I notice no mention of Tiered Storage (KIP-405). Does that complicate the > design, especially when fetching historical data? It would be good to have > at least one sentence mentioning it, even if it doesn't impact it. Right > now I'm unsure if it was considered. > > 2. SSO initialized to the latest offset > > > "By default, the SSO for each share-partition is initialized to the > latest offset for the corresponding topic-partitions." > > Have we considered allowing this to be configurable to latest/earliest? > This would be consistent with the auto.offset.reset config of vanilla > consumer groups. > Thinking from a user's perspective, it sounds valid to want to start from > the start of a topic when starting a share group. Historical processing > comes to mind > > 3. Durable Storage > > The KIP mentions that "The cluster records this information durably", which > implies that it saves it somewhere. Does the ShareCoordinator have its own > topic? Would it be compacted? > > In particular, I am interested in what such a topic's retention would be > like. The vanilla consumer offsets topic has some special retention > semantics (KIP-211) where we start counting the retention after the > consumer group becomes empty (inactive) - the default being 7 days. Need to > make sure the retention here isn't too short either, as the offsets topic > originally had 24 hours of retention and that proved problematic. > > In general, some extra detail about the persistence would be greatly > appreciated! > > 4. Batch Acknowledgement > > > "In the situation where some records in a batch have been released or > rejected separately, subsequent fetches of those records are more likely to > have gaps." > > Can we expand a bit more on this edge case? I am interested in learning > what gets returned on subsequent fetch requests. > In particular, - how does this work with compression? As far as I remember, > we can compress the whole batch there, which might make individual record > filtering tricky. > > 5. Member Management > > How is consumer group member management handled? I didn't see any specific > mention - is it the same as a vanilla group? > In particular - how will bad consumers be handled? > > I guess I see two cases: > 1. bad consumer that doesn't even heartbeat > 2. bad consumer that heartbeats well but for some reason every message > processing times out. e.g imagine it was network partitioned from some > third-party system that is a critical part of its message processing loop > > One evident problem I can foresee in production systems is one (or a few) > slow consumer applications bringing the SSO/SEO advancement down to a crawl. > Imagine an example where the same consumer app always hits the timeout > limit - what would the behavior be in such a case? Do we keep that consumer > app indefinitely (if so, do we run the risk of having it invalidate > completely valid messages)? Are there any equivalents to the consumer group > rebalances which fence off such bad consumers? > > 6. Processing Semantics (exactly once) > > > The delivery counts are only maintained approximately and the Acquired > state is not persisted. > > Does this introduce the risk of zombie consumers on share-partition-leader > failure? i.e restarting and giving another consumer the acquired state for > the same record > > I notice that the KIP says: > > Finally, this KIP does not include support for acknowledging delivery > using transactions for exactly-once semantics. > at the very end. It would be helpful to address this earlier in the > example, as one of the key points. And it would be good to be clearer on > what the processing semantics are. They seem to be *at-least-once* to me. > > > 7. nit: Acronyms > > I feel like SSO and SEO may be bad acronyms. Googling "Kafka SEO" is bound > to return weird results. > What do we think about the tradeoff of using more-unique acronyms (like > SGEO, SSGO) at the expense of one extra letter? > > Again - thanks for working on this! I think it's a great initiative. I'm > excited to see us perfect this proposal and enable a brand new use case in > Kafka! > > Best, > Stanislav > > > > On Mon, May 15, 2023 at 2:55 PM Andrew Schofield <andrew_schofi...@live.com> > wrote: > > > Hi, > > I would like to start a discussion thread on KIP-932: Queues for Kafka. > > This KIP proposes an alternative to consumer groups to enable cooperative > > consumption by consumers without partition assignment. You end up with > > queue semantics on top of regular Kafka topics, with per-message > > acknowledgement and automatic handling of messages which repeatedly fail to > > be processed. > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka > > > > Please take a look and let me know what you think. > > > > Thanks. > > Andrew > > > > -- > Best, > Stanislav
