[jira] [Created] (KAFKA-14367) Introduce `GroupCoordinator` interface

[David Jacot (Jira)]

David Jacot created KAFKA-14367:
---

 Summary: Introduce `GroupCoordinator` interface
 Key: KAFKA-14367
 URL: https://issues.apache.org/jira/browse/KAFKA-14367
 Project: Kafka
  Issue Type: Sub-task
Reporter: David Jacot
Assignee: David Jacot


The goal is to introduce a new GroupCoordinator interface and to convert the 
existing coordinator to use it. The new coordinator will use it later on.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] KIP-852 Optimize calculation of size for log in remote tier

[Divij Vaidya]

Hey Jun

Thank you for your comments.

*1. "RLMM implementor could ensure that listRemoteLogSegments() is fast"*
This would be ideal but pragmatically, it is difficult to ensure that
listRemoteLogSegments() is fast. This is because of the possibility of a
large number of segments (much larger than what Kafka currently handles
with local storage today) would make it infeasible to adopt strategies such
as local caching to improve the performance of listRemoteLogSegments. Apart
from caching (which won't work due to size limitations) I can't think of
other strategies which may eliminate the need for IO
operations proportional to the number of total segments. Please advise if
you have something in mind.

2.  "*If the size exceeds the retention size, we need to determine the
subset of segments to delete to bring the size within the retention limit.
Do we need to call RemoteLogMetadataManager.listRemoteLogSegments() to
determine that?"*
Yes, we need to call listRemoteLogSegments() to determine which segments
should be deleted. But there is a difference with the use case we are
trying to optimize with this KIP. To determine the subset of segments which
would be deleted, we only read metadata for segments which would be deleted
via the listRemoteLogSegments(). But to determine the totalLogSize, which
is required every time retention logic based on size executes, we read
metadata of *all* the segments in remote storage. Hence, the number of
results returned by *RemoteLogMetadataManager.listRemoteLogSegments() *is
different when we are calculating totalLogSize vs. when we are determining
the subset of segments to delete.

3.
*"Also, what about time-based retention? To make that efficient, do we need
to make some additional interface changes?"*No. Note that time complexity
to determine the segments for retention is different for time based vs.
size based. For time based, the time complexity is a function of the number
of segments which are "eligible for deletion" (since we only read metadata
for segments which would be deleted) whereas in size based retention, the
time complexity is a function of "all segments" available in remote storage
(metadata of all segments needs to be read to calculate the total size). As
you may observe, this KIP will bring the time complexity for both time
based retention & size based retention to the same function.

4. Also, please note that this new API introduced in this KIP also enables
us to provide a metric for total size of data stored in remote storage.
Without the API, calculation of this metric will become very expensive with
*listRemoteLogSegments().*
I understand that your motivation here is to avoid polluting the interface
with optimization specific APIs and I will agree with that goal. But I
believe that this new API proposed in the KIP brings in significant
improvement and there is no other work around available to achieve the same
performance.

Regards,
Divij Vaidya



On Tue, Nov 8, 2022 at 12:12 AM Jun Rao  wrote:

> Hi, Divij,
>
> Thanks for the KIP. Sorry for the late reply.
>
> The motivation of the KIP is to improve the efficiency of size based
> retention. I am not sure the proposed changes are enough. For example, if
> the size exceeds the retention size, we need to determine the subset of
> segments to delete to bring the size within the retention limit. Do we need
> to call RemoteLogMetadataManager.listRemoteLogSegments() to determine that?
> Also, what about time-based retention? To make that efficient, do we need
> to make some additional interface changes?
>
> An alternative approach is for the RLMM implementor to make sure
> that RemoteLogMetadataManager.listRemoteLogSegments() is fast (e.g., with
> local caching). This way, we could keep the interface simple. Have we
> considered that?
>
> Thanks,
>
> Jun
>
> On Wed, Sep 28, 2022 at 6:28 AM Divij Vaidya 
> wrote:
>
> > Hey folks
> >
> > Does anyone else have any thoughts on this before I propose this for a
> > vote?
> >
> > --
> > Divij Vaidya
> >
> >
> >
> > On Mon, Sep 5, 2022 at 12:57 PM Satish Duggana  >
> > wrote:
> >
> > > Thanks for the KIP Divij!
> > >
> > > This is a nice improvement to avoid recalculation of size. Customized
> > RLMMs
> > > can implement the best possible approach by caching or maintaining the
> > size
> > > in an efficient way. But this is not a big concern for the default
> topic
> > > based RLMM as mentioned in the KIP.
> > >
> > > ~Satish.
> > >
> > > On Wed, 13 Jul 2022 at 18:48, Divij Vaidya 
> > > wrote:
> > >
> > > > Thank you for your review Luke.
> > > >
> > > > > Reg: is that would the new `RemoteLogSizeBytes` metric be a
> > performance
> > > > overhead? Although we move the calculation to a seperate API, we
> still
> > > > can't assume users will implement a light-weight method, right?
> > > >
> > > > This metric would be logged using the information that is already
> being
> > > > calculated for handling remote retention logic, hence, no additional
> > work
> > > > is re

Re: [DISCUSS] KIP-875: First-class offsets support in Kafka Connect

[Yash Mayya]

Hi Chris,

Thanks for the response and the explanations, I think you've answered
pretty much all the questions I had meticulously!


> if something goes wrong while resetting offsets, there's no
> immediate impact--the connector will still be in the STOPPED
>  state. The REST response for requests to reset the offsets
> will clearly call out that the operation has failed, and if necessary,
> we can probably also add a scary-looking warning message
> stating that we can't guarantee which offsets have been successfully
>  wiped and which haven't. Users can query the exact offsets of
> the connector at this point to determine what will happen if/what they
> resume it. And they can repeat attempts to reset the offsets as many
>  times as they'd like until they get back a 2XX response, indicating
> that it's finally safe to resume the connector. Thoughts?

Yeah, I agree, the case that I mentioned earlier where a user would try to
resume a stopped connector after a failed offset reset attempt without
knowing that the offset reset attempt didn't fail cleanly is probably just
an extreme edge case. I think as long as the response is verbose enough and
self explanatory, we should be fine.

Another question that I had was behavior w.r.t sink connector offset resets
when there are zombie tasks/workers in the Connect cluster - the KIP
mentions that for sink connectors offset resets will be done by deleting
the consumer group. However, if there are zombie tasks which are still able
to communicate with the Kafka cluster that the sink connector is consuming
from, I think the consumer group will automatically get re-created and the
zombie task may be able to commit offsets for the partitions that it is
consuming from?

Thanks,
Yash


On Fri, Nov 4, 2022 at 10:27 PM Chris Egerton 
wrote:

> Hi Yash,
>
> Thanks again for your thoughts! Responses to ongoing discussions inline
> (easier to track context than referencing comment numbers):
>
> > However, this then leads me to wonder if we can make that explicit by
> including "connect" or "connector" in the higher level field names? Or do
> you think this isn't required given that we're talking about a Connect
> specific REST API in the first place?
>
> I think "partition" and "offset" are fine as field names but I'm not hugely
> opposed to adding "connector " as a prefix to them; would be interested in
> others' thoughts.
>
> > I'm not sure I followed why the unresolved writes to the config topic
> would be an issue - wouldn't the delete offsets request be added to the
> herder's request queue and whenever it is processed, we'd anyway need to
> check if all the prerequisites for the request are satisfied?
>
> Some requests are handled in multiple steps. For example, deleting a
> connector (1) adds a request to the herder queue to write a tombstone to
> the config topic (or, if the worker isn't the leader, forward the request
> to the leader). (2) Once that tombstone is picked up, (3) a rebalance
> ensues, and then after it's finally complete, (4) the connector and its
> tasks are shut down. I probably could have used better terminology, but
> what I meant by "unresolved writes to the config topic" was a case in
> between steps (2) and (3)--where the worker has already read that tombstone
> from the config topic and knows that a rebalance is pending, but hasn't
> begun participating in that rebalance yet. In the DistributedHerder class,
> this is done via the `checkRebalanceNeeded` method.
>
> > We can probably revisit this potential deprecation [of the PAUSED state]
> in the future based on user feedback and how the adoption of the new
> proposed stop endpoint looks like, what do you think?
>
> Yeah, revisiting in the future seems reasonable. 👍
>
> And responses to new comments here:
>
> 8. Yep, we'll start tracking offsets by connector. I don't believe this
> should be too difficult, and suspect that the only reason we track raw byte
> arrays instead of pre-deserializing offset topic information into something
> more useful is because Connect originally had pluggable internal
> converters. Now that we're hardcoded to use the JSON converter it should be
> fine to track offsets on a per-connector basis as they're read from the
> offsets topic.
>
> 9. I'm hesitant to introduce this type of feature right now because of all
> of the gotchas that would come with it. In security-conscious environments,
> it's possible that a sink connector's principal may have access to the
> consumer group used by the connector, but the worker's principal may not.
> There's also the case where source connectors have separate offsets topics,
> or sink connectors have overridden consumer group IDs, or sink or source
> connectors work against a different Kafka cluster than the one that their
> worker uses. Overall, I'd rather provide a single API that works in all
> cases rather than risk confusing and alienating users by trying to make
> their lives easier in a subset of cases.
>
> 10. Hmm... I don't 

[jira] [Created] (KAFKA-14368) Add an offset write REST API to Kafka Connect

[Yash Mayya (Jira)]

Yash Mayya created KAFKA-14368:
--

 Summary: Add an offset write REST API to Kafka Connect
 Key: KAFKA-14368
 URL: https://issues.apache.org/jira/browse/KAFKA-14368
 Project: Kafka
  Issue Type: New Feature
Reporter: Yash Mayya
Assignee: Yash Mayya


[KIP-875|https://cwiki.apache.org/confluence/display/KAFKA/KIP-875%3A+First-class+offsets+support+in+Kafka+Connect]
 for https://issues.apache.org/jira/browse/KAFKA-4107 proposes to add an offset 
reset API which will allow resetting the offsets for source and sink connectors 
so that they can consume from the beginning of the stream. However, an offset 
API to write arbitrary offsets would also be useful for certain connectors in 
order to go back in time but not to the beginning, or to skip some problematic 
record and move forward. Based on the discussion thread for KIP-875 
[here|https://lists.apache.org/thread/m5bklnh5w4mwr9nbzrmfk0pftpxfjd02], it was 
determined that this could be done through a follow-up KIP if/when KIP-875 is 
adopted.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] KIP-872: Add Serializer#serializeToByteBuffer() to reduce memory copying

[ShunKang Lin]

Thanks John!

I will consider your opinion, I’ve been busy at work lately.

Best,
ShunKang

John Roesler 于2022年11月6日 周日23:22写道：

> Thanks for the reply, ShunKang!
>
> You’re absolutely right, we should not change the behavior of the existing
> method.
>
> Regarding the new method, I was thinking that this is a good opportunity
> to correct what seems to be strange semantics in the original one. If we
> keep the same semantics and want to correct it later, we’ll be forced to
> introduce yet another method later. This especially makes sense if we’re
> thinking of deprecating the original method. But if you think it’s better
> to keep it the way it is, I’m fine with it.
>
> I have no other comments.
>
> Thanks again for the KIP,
> John
>
> On Sat, Nov 5, 2022, at 11:59, ShunKang Lin wrote:
> > Hi John,
> >
> > Thanks for your comments!
> >
> > For your first question, I see some unit test cases that give us a
> > ByteBuffer not set to read before calling
> > `ByteBufferSerializer#serialize(String, ByteBuffer)`, e.g.
> > `ArticleSerializer`, `AugmentedArticleSerializer`,
> > `AugmentedCommentSerializer` and `CommentSerializer`. If we don't flip
> the
> > ByteBuffer inside the `ByteBufferSerializer#serialize(String,
> ByteBuffer)`
> > it will break user code using `ByteBufferSerializer#serialize(String,
> > ByteBuffer)`, and if we don't flip the ByteBuffer inside
> > the `ByteBufferSerializer#serializeToByteBuffer(String, ByteBuffer)`, it
> > will be even more strange to the user, because
> > `ByteBufferSerializer#serialize(String, ByteBuffer)` and
> > `ByteBufferSerializer#serializeToByteBuffer(String, ByteBuffer)` require
> > users use the ByteBufferSerializer in two different ways. So if we think
> of
> > `ByteBufferSerialize#serializeToByteBuffer(String, ByteBuffer)` as
> setting
> > up a ByteBuffer to read later, is it more acceptable?
> >
> > For your second question, I plan to ultimately replace byte[] with
> > ByteBuffer, I will document the intent in your KIP and JavaDocs later.
> >
> > I will clarify that if a Serializer implements the new method, then the
> old
> > one will never be called.
> >
> > Best,
> > ShunKang
> >
> > John Roesler  于2022年11月4日周五 22:42写道：
> >
> >> Hi ShunKang,
> >>
> >> Thanks for the KIP!
> >>
> >> I’ve been wanting to transition toward byte buffers for a while, so this
> >> is a nice start.
> >>
> >> I thought it was a bit weird to flip the buffer inside the serializer,
> but
> >> I see the existing one already does that. I would have thought it would
> >> make more sense for the caller to give us a buffer already set up for
> >> reading. Do you think it makes sense to adopt this pattern for the new
> >> method?
> >>
> >> Do you plan to keep the new methods as optional indefinitely, or do you
> >> plan to ultimately replace byte[] with ByteBuffer? If it’s the latter,
> then
> >> it would be good to document the intent in your KIP and JavaDocs.
> >>
> >> It would be good to clarify that if a Serializer implements the new
> >> method, then the old one will never be called. That way, implementations
> >> can just throw an exception on that method instead of implementing both.
> >>
> >> Thanks again!
> >> -John
> >>
> >> On Wed, Nov 2, 2022, at 20:14, ShunKang Lin wrote:
> >> > Bump this thread again : )
> >> >
> >> > ShunKang Lin 于2022年9月25日 周日23:59写道：
> >> >
> >> >> Hi all, I'd like to start a new discussion thread on KIP-872 (Kafka
> >> >> Client) which proposes that add Serializer#serializeToByteBuffer() to
> >> >> reduce memory copying.
> >> >>
> >> >> KIP:
> >> >>
> >>
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=228495828
> >> >> Thanks, ShunKang
> >> >>
> >>
>

Re: [DISCUSS] KIP-866 ZooKeeper to KRaft Migration

[David Arthur]

Thanks for the discussion everyone, I'm going to move ahead with the
vote for this KIP.

-David

On Thu, Nov 3, 2022 at 1:20 PM Jun Rao  wrote:
>
> Hi, David,
>
> Thanks for the reply.
>
> 20/21 Yes, but separate listeners are optional. It's possible for the nodes
> to use a single port for both client and server side communications.
>
> Thanks,
>
> Jun
>
> On Thu, Nov 3, 2022 at 9:59 AM David Arthur
>  wrote:
>
> > 20/21, in combined mode we still have a separate listener for the
> > controller APIs, e.g.,
> >
> > listeners=PLAINTEXT://:9092,CONTROLLER://:9093
> >
> > inter.broker.listener.name=PLAINTEXT
> >
> > controller.listener.names=CONTROLLER
> >
> > advertised.listeners=PLAINTEXT://localhost:9092
> >
> >
> >
> > Clients still talk to the broker through the advertised listener, and only
> > brokers and other controllers will communicate over the controller
> > listener.
> >
> > 40. Sounds good, I updated the KIP
> >
> > Thanks!
> > David
> >
> >
> >
> > On Thu, Nov 3, 2022 at 12:14 PM Jun Rao  wrote:
> >
> > > Hi, David,
> > >
> > > Thanks for the reply.
> > >
> > > 20/21. When KRaft runs in the combined mode, does a controller know
> > whether
> > > an ApiRequest is from a client or another broker?
> > >
> > > 40. Adding a "None" state sounds reasonable.
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > > On Thu, Nov 3, 2022 at 8:39 AM David Arthur
> > >  wrote:
> > >
> > > > Jun,
> > > >
> > > > 20/21 If we use a tagged field, then I don't think clients need to be
> > > > concerned with it, right?. In ApiVersionsResponse sent by brokers to
> > > > clients, this field would be omitted. Clients can't connect directly to
> > > the
> > > > KRaft controller nodes. Also, we have a precedent of sending controller
> > > > node state between controllers through ApiVersions ("metadata.version"
> > > > feature), so I think it fits well.
> > > >
> > > > 40. For new KRaft clusters, we could add another state to indicate it
> > was
> > > > not migrated from ZK, but created as a KRaft cluster. Maybe
> > > > "kafka.controller:type=KafkaController,name=MigrationState" => "None" ?
> > > We
> > > > could also omit that metric for unmigrated clusters, but I'm not a fan
> > of
> > > > using the absence of a value to signal something.
> > > >
> > > > -
> > > >
> > > > Akhilesh, thanks for reviewing the KIP!
> > > >
> > > > 1. MigrationState and MetadataType are mostly the same on the
> > controller,
> > > > but we do have the "MigratingZkData" state that seems useful to report
> > > as a
> > > > metric. Aside from looking at logs, observing the controller in this
> > > state
> > > > is the only way to see how long its taking to copy data from ZK.
> > > >
> > > > "KRaftMode" instead of "MigrationFinalized" is similar to Jun's
> > question
> > > > about non-migrated clusters. I think it's useful to have a distinct
> > > > MigrationState for clusters that have been migrated and those that were
> > > > never migrated. This does mean we'll report the MigrationState long
> > after
> > > > the migration is complete, but we can drop these metrics in 4.0 once ZK
> > > is
> > > > removed.
> > > >
> > > > 2. The "ZkMigrationReady" will indicate that the controller has
> > > > "kafka.metadata.migration.enable" _and_ the ZK configs set. We need
> > some
> > > > way to indicate that the whole quorum is correctly configured to handle
> > > the
> > > > migration so we don't failover to a controller that's not configured
> > for
> > > > ZK. Did I understand your question correctly?
> > > >
> > > > 3. Yea, good idea. While the KRaft controller has
> > > > MigrationState=MigrationIneligible, we could also report
> > > >
> > >
> > "kafka.controller:type=KafkaController,name=MigrationInelgibleBrokerCount".
> > > > It might be useful to report ineligible controllers as well since that
> > > can
> > > > prevent the migration from starting.
> > > >
> > > > 4. I think I covered this in "Incompatible Brokers". We effectively
> > fence
> > > > these brokers by not sending them metadata RPCs.
> > > >
> > > > Thanks!
> > > > David
> > > >
> > > >
> > > > On Tue, Nov 1, 2022 at 3:58 PM Akhilesh Chaganti <
> > akhilesh@gmail.com
> > > >
> > > > wrote:
> > > >
> > > > > Hi David,
> > > > >
> > > > >
> > > > > Thanks for the KIP. I have some questions/suggestions.
> > > > >
> > > > >
> > > > > 1) I see two new metrics:
> > > > > kafka.controller:type=KafkaController,name=MetadataType and
> > > > > kafka.controller:type=KafkaController,name=MigrationState. Won't the
> > > > second
> > > > > metric already cover the cases of the first metric? Also, instead of
> > > > > MigrationFinalized, we could directly say the state is KRaftMode. So
> > we
> > > > can
> > > > > use the same value for default KRaft clusters.
> > > > >
> > > > >
> > > > > 2) ZkMigrationReady in ApiVersionsResponse from KRaft Controller. By
> > > > > default, we plan to start the Controller quorum in "
> > > > > *kafka.metadata.migration.enable*" config set to true. Then do we
> > 

[VOTE] KIP-866 ZooKeeper to KRaft Migration

[David Arthur]

Hello everyone, I'd like to start the vote on KIP-866.

https://cwiki.apache.org/confluence/display/KAFKA/KIP-866+ZooKeeper+to+KRaft+Migration

Thanks!
David Arthur

Re: [DISCUSS] KIP-866 ZooKeeper to KRaft Migration

[Jun Rao]

Hi, David,

I am not sure that we are fully settled on the following.

20/21. Since separate listeners are optional, it seems that the broker
can't distinguish between ApiVersion requests coming from the client or
other brokers. This means the clients will get ZkMigrationReady in the
ApiVersion response, which is weird.

Thanks,

Jun

On Tue, Nov 8, 2022 at 7:18 AM David Arthur  wrote:

> Thanks for the discussion everyone, I'm going to move ahead with the
> vote for this KIP.
>
> -David
>
> On Thu, Nov 3, 2022 at 1:20 PM Jun Rao  wrote:
> >
> > Hi, David,
> >
> > Thanks for the reply.
> >
> > 20/21 Yes, but separate listeners are optional. It's possible for the
> nodes
> > to use a single port for both client and server side communications.
> >
> > Thanks,
> >
> > Jun
> >
> > On Thu, Nov 3, 2022 at 9:59 AM David Arthur
> >  wrote:
> >
> > > 20/21, in combined mode we still have a separate listener for the
> > > controller APIs, e.g.,
> > >
> > > listeners=PLAINTEXT://:9092,CONTROLLER://:9093
> > >
> > > inter.broker.listener.name=PLAINTEXT
> > >
> > > controller.listener.names=CONTROLLER
> > >
> > > advertised.listeners=PLAINTEXT://localhost:9092
> > >
> > >
> > >
> > > Clients still talk to the broker through the advertised listener, and
> only
> > > brokers and other controllers will communicate over the controller
> > > listener.
> > >
> > > 40. Sounds good, I updated the KIP
> > >
> > > Thanks!
> > > David
> > >
> > >
> > >
> > > On Thu, Nov 3, 2022 at 12:14 PM Jun Rao 
> wrote:
> > >
> > > > Hi, David,
> > > >
> > > > Thanks for the reply.
> > > >
> > > > 20/21. When KRaft runs in the combined mode, does a controller know
> > > whether
> > > > an ApiRequest is from a client or another broker?
> > > >
> > > > 40. Adding a "None" state sounds reasonable.
> > > >
> > > > Thanks,
> > > >
> > > > Jun
> > > >
> > > > On Thu, Nov 3, 2022 at 8:39 AM David Arthur
> > > >  wrote:
> > > >
> > > > > Jun,
> > > > >
> > > > > 20/21 If we use a tagged field, then I don't think clients need to
> be
> > > > > concerned with it, right?. In ApiVersionsResponse sent by brokers
> to
> > > > > clients, this field would be omitted. Clients can't connect
> directly to
> > > > the
> > > > > KRaft controller nodes. Also, we have a precedent of sending
> controller
> > > > > node state between controllers through ApiVersions
> ("metadata.version"
> > > > > feature), so I think it fits well.
> > > > >
> > > > > 40. For new KRaft clusters, we could add another state to indicate
> it
> > > was
> > > > > not migrated from ZK, but created as a KRaft cluster. Maybe
> > > > > "kafka.controller:type=KafkaController,name=MigrationState" =>
> "None" ?
> > > > We
> > > > > could also omit that metric for unmigrated clusters, but I'm not a
> fan
> > > of
> > > > > using the absence of a value to signal something.
> > > > >
> > > > > -
> > > > >
> > > > > Akhilesh, thanks for reviewing the KIP!
> > > > >
> > > > > 1. MigrationState and MetadataType are mostly the same on the
> > > controller,
> > > > > but we do have the "MigratingZkData" state that seems useful to
> report
> > > > as a
> > > > > metric. Aside from looking at logs, observing the controller in
> this
> > > > state
> > > > > is the only way to see how long its taking to copy data from ZK.
> > > > >
> > > > > "KRaftMode" instead of "MigrationFinalized" is similar to Jun's
> > > question
> > > > > about non-migrated clusters. I think it's useful to have a distinct
> > > > > MigrationState for clusters that have been migrated and those that
> were
> > > > > never migrated. This does mean we'll report the MigrationState long
> > > after
> > > > > the migration is complete, but we can drop these metrics in 4.0
> once ZK
> > > > is
> > > > > removed.
> > > > >
> > > > > 2. The "ZkMigrationReady" will indicate that the controller has
> > > > > "kafka.metadata.migration.enable" _and_ the ZK configs set. We need
> > > some
> > > > > way to indicate that the whole quorum is correctly configured to
> handle
> > > > the
> > > > > migration so we don't failover to a controller that's not
> configured
> > > for
> > > > > ZK. Did I understand your question correctly?
> > > > >
> > > > > 3. Yea, good idea. While the KRaft controller has
> > > > > MigrationState=MigrationIneligible, we could also report
> > > > >
> > > >
> > >
> "kafka.controller:type=KafkaController,name=MigrationInelgibleBrokerCount".
> > > > > It might be useful to report ineligible controllers as well since
> that
> > > > can
> > > > > prevent the migration from starting.
> > > > >
> > > > > 4. I think I covered this in "Incompatible Brokers". We effectively
> > > fence
> > > > > these brokers by not sending them metadata RPCs.
> > > > >
> > > > > Thanks!
> > > > > David
> > > > >
> > > > >
> > > > > On Tue, Nov 1, 2022 at 3:58 PM Akhilesh Chaganti <
> > > akhilesh@gmail.com
> > > > >
> > > > > wrote:
> > > > >
> > > > > > Hi David,
> > > > > >
> > > > > >
> > > > > > Thanks for the KIP. I have some questions/sug

Re: [DISCUSS] KIP-875: First-class offsets support in Kafka Connect

[Chris Egerton]

Hi Yash,

Good question! This is actually a subtle source of asymmetry in the current
proposal. Requests to delete a consumer group with active members will
fail, so if there are zombie sink tasks that are still communicating with
Kafka, offset reset requests for that connector will also fail. It is
possible to use an admin client to remove all active members from the group
and then delete the group. However, this solution isn't as complete as the
zombie fencing that we can perform for exactly-once source tasks, since
removing consumers from a group doesn't prevent them from immediately
rejoining the group, which would either cause the group deletion request to
fail (if they rejoin before the group is deleted), or recreate the group
(if they rejoin after the group is deleted).

For ease of implementation, I'd prefer to leave the asymmetry in the API
for now and fail fast and clearly if there are still consumers active in
the sink connector's group. We can try to detect this case and provide a
helpful error message to the user explaining why the offset reset request
has failed and some steps they can take to try to resolve things (wait for
slow task shutdown to complete, restart zombie workers and/or workers with
blocked tasks on them). In the future we can possibly even revisit KIP-611
[1] or something like it to provide better insight into zombie tasks on a
worker so that it's easier to find which tasks have been abandoned but are
still running.

Let me know what you think; this is an important point to call out and if
we can reach some consensus on how to handle sink connector offset resets
w/r/t zombie tasks, I'll update the KIP with the details.

[1] -
https://cwiki.apache.org/confluence/display/KAFKA/KIP-611%3A+Improved+Handling+of+Abandoned+Connectors+and+Tasks

Cheers,

Chris

On Tue, Nov 8, 2022 at 8:00 AM Yash Mayya  wrote:

> Hi Chris,
>
> Thanks for the response and the explanations, I think you've answered
> pretty much all the questions I had meticulously!
>
>
> > if something goes wrong while resetting offsets, there's no
> > immediate impact--the connector will still be in the STOPPED
> >  state. The REST response for requests to reset the offsets
> > will clearly call out that the operation has failed, and if necessary,
> > we can probably also add a scary-looking warning message
> > stating that we can't guarantee which offsets have been successfully
> >  wiped and which haven't. Users can query the exact offsets of
> > the connector at this point to determine what will happen if/what they
> > resume it. And they can repeat attempts to reset the offsets as many
> >  times as they'd like until they get back a 2XX response, indicating
> > that it's finally safe to resume the connector. Thoughts?
>
> Yeah, I agree, the case that I mentioned earlier where a user would try to
> resume a stopped connector after a failed offset reset attempt without
> knowing that the offset reset attempt didn't fail cleanly is probably just
> an extreme edge case. I think as long as the response is verbose enough and
> self explanatory, we should be fine.
>
> Another question that I had was behavior w.r.t sink connector offset resets
> when there are zombie tasks/workers in the Connect cluster - the KIP
> mentions that for sink connectors offset resets will be done by deleting
> the consumer group. However, if there are zombie tasks which are still able
> to communicate with the Kafka cluster that the sink connector is consuming
> from, I think the consumer group will automatically get re-created and the
> zombie task may be able to commit offsets for the partitions that it is
> consuming from?
>
> Thanks,
> Yash
>
>
> On Fri, Nov 4, 2022 at 10:27 PM Chris Egerton 
> wrote:
>
> > Hi Yash,
> >
> > Thanks again for your thoughts! Responses to ongoing discussions inline
> > (easier to track context than referencing comment numbers):
> >
> > > However, this then leads me to wonder if we can make that explicit by
> > including "connect" or "connector" in the higher level field names? Or do
> > you think this isn't required given that we're talking about a Connect
> > specific REST API in the first place?
> >
> > I think "partition" and "offset" are fine as field names but I'm not
> hugely
> > opposed to adding "connector " as a prefix to them; would be interested
> in
> > others' thoughts.
> >
> > > I'm not sure I followed why the unresolved writes to the config topic
> > would be an issue - wouldn't the delete offsets request be added to the
> > herder's request queue and whenever it is processed, we'd anyway need to
> > check if all the prerequisites for the request are satisfied?
> >
> > Some requests are handled in multiple steps. For example, deleting a
> > connector (1) adds a request to the herder queue to write a tombstone to
> > the config topic (or, if the worker isn't the leader, forward the request
> > to the leader). (2) Once that tombstone is picked up, (3) a rebalance
> > ensues, and then after it's fi

[jira] [Created] (KAFKA-14369) Docs - KRAFT controller authentication example

[Domenic Bove (Jira)]

Domenic Bove created KAFKA-14369:


 Summary: Docs - KRAFT controller authentication example
 Key: KAFKA-14369
 URL: https://issues.apache.org/jira/browse/KAFKA-14369
 Project: Kafka
  Issue Type: Bug
  Components: docs
Affects Versions: 3.3.1
Reporter: Domenic Bove


The [Kafka Listener docs 
|https://kafka.apache.org/documentation/#listener_configuration]mention how to 
handle kafka protocols (other than PLAINTEXT) on the KRAFT controller listener, 
but it is not a working example and I found that I was missing this property: 
`sasl.mechanism.controller.protocol` when attempting to do SASL_PLAINTEXT on 
the controller listener. I see that property here: 
[https://kafka.apache.org/documentation/#brokerconfigs_sasl.mechanism.controller.protocol]

But nowhere else. 

I wonder if a complete working example would be better. Here are my working 
configs for sasl plain on the controller

 
{code:java}
process.roles=controller
listeners=CONTROLLER://:9093 
node.id=1
controller.quorum.voters=1@localhost:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT
listener.name.controller.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
 required username="admin" password="admin-secret" user_admin="admin-secret" 
user_alice="alice-secret";
listener.name.controller.sasl.enabled.mechanisms=PLAIN
listener.name.controller.sasl.mechanism=PLAIN
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.controller.protocol=PLAIN{code}
 

Or maybe just a callout of that property in the existing docs



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [DISCUSS] KIP-866 ZooKeeper to KRaft Migration

[David Arthur]

Ah, sorry about that, you're right. Since we won't support ZK
migrations in combined mode, is this issue avoided?

Essentially, we would only set ZkMigrationReady in ApiVersionsResponse if

* process.roles=controller
* kafka.metadata.migration.enabled=true

Also, the following would be an invalid config that should prevent startup:

* process.roles=broker,controller
* kafka.metadata.migration.enabled=true

Does this seem reasonable?

Thanks!
David

On Tue, Nov 8, 2022 at 11:12 AM Jun Rao  wrote:
>
> Hi, David,
>
> I am not sure that we are fully settled on the following.
>
> 20/21. Since separate listeners are optional, it seems that the broker
> can't distinguish between ApiVersion requests coming from the client or
> other brokers. This means the clients will get ZkMigrationReady in the
> ApiVersion response, which is weird.
>
> Thanks,
>
> Jun
>
> On Tue, Nov 8, 2022 at 7:18 AM David Arthur  wrote:
>
> > Thanks for the discussion everyone, I'm going to move ahead with the
> > vote for this KIP.
> >
> > -David
> >
> > On Thu, Nov 3, 2022 at 1:20 PM Jun Rao  wrote:
> > >
> > > Hi, David,
> > >
> > > Thanks for the reply.
> > >
> > > 20/21 Yes, but separate listeners are optional. It's possible for the
> > nodes
> > > to use a single port for both client and server side communications.
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > > On Thu, Nov 3, 2022 at 9:59 AM David Arthur
> > >  wrote:
> > >
> > > > 20/21, in combined mode we still have a separate listener for the
> > > > controller APIs, e.g.,
> > > >
> > > > listeners=PLAINTEXT://:9092,CONTROLLER://:9093
> > > >
> > > > inter.broker.listener.name=PLAINTEXT
> > > >
> > > > controller.listener.names=CONTROLLER
> > > >
> > > > advertised.listeners=PLAINTEXT://localhost:9092
> > > >
> > > >
> > > >
> > > > Clients still talk to the broker through the advertised listener, and
> > only
> > > > brokers and other controllers will communicate over the controller
> > > > listener.
> > > >
> > > > 40. Sounds good, I updated the KIP
> > > >
> > > > Thanks!
> > > > David
> > > >
> > > >
> > > >
> > > > On Thu, Nov 3, 2022 at 12:14 PM Jun Rao 
> > wrote:
> > > >
> > > > > Hi, David,
> > > > >
> > > > > Thanks for the reply.
> > > > >
> > > > > 20/21. When KRaft runs in the combined mode, does a controller know
> > > > whether
> > > > > an ApiRequest is from a client or another broker?
> > > > >
> > > > > 40. Adding a "None" state sounds reasonable.
> > > > >
> > > > > Thanks,
> > > > >
> > > > > Jun
> > > > >
> > > > > On Thu, Nov 3, 2022 at 8:39 AM David Arthur
> > > > >  wrote:
> > > > >
> > > > > > Jun,
> > > > > >
> > > > > > 20/21 If we use a tagged field, then I don't think clients need to
> > be
> > > > > > concerned with it, right?. In ApiVersionsResponse sent by brokers
> > to
> > > > > > clients, this field would be omitted. Clients can't connect
> > directly to
> > > > > the
> > > > > > KRaft controller nodes. Also, we have a precedent of sending
> > controller
> > > > > > node state between controllers through ApiVersions
> > ("metadata.version"
> > > > > > feature), so I think it fits well.
> > > > > >
> > > > > > 40. For new KRaft clusters, we could add another state to indicate
> > it
> > > > was
> > > > > > not migrated from ZK, but created as a KRaft cluster. Maybe
> > > > > > "kafka.controller:type=KafkaController,name=MigrationState" =>
> > "None" ?
> > > > > We
> > > > > > could also omit that metric for unmigrated clusters, but I'm not a
> > fan
> > > > of
> > > > > > using the absence of a value to signal something.
> > > > > >
> > > > > > -
> > > > > >
> > > > > > Akhilesh, thanks for reviewing the KIP!
> > > > > >
> > > > > > 1. MigrationState and MetadataType are mostly the same on the
> > > > controller,
> > > > > > but we do have the "MigratingZkData" state that seems useful to
> > report
> > > > > as a
> > > > > > metric. Aside from looking at logs, observing the controller in
> > this
> > > > > state
> > > > > > is the only way to see how long its taking to copy data from ZK.
> > > > > >
> > > > > > "KRaftMode" instead of "MigrationFinalized" is similar to Jun's
> > > > question
> > > > > > about non-migrated clusters. I think it's useful to have a distinct
> > > > > > MigrationState for clusters that have been migrated and those that
> > were
> > > > > > never migrated. This does mean we'll report the MigrationState long
> > > > after
> > > > > > the migration is complete, but we can drop these metrics in 4.0
> > once ZK
> > > > > is
> > > > > > removed.
> > > > > >
> > > > > > 2. The "ZkMigrationReady" will indicate that the controller has
> > > > > > "kafka.metadata.migration.enable" _and_ the ZK configs set. We need
> > > > some
> > > > > > way to indicate that the whole quorum is correctly configured to
> > handle
> > > > > the
> > > > > > migration so we don't failover to a controller that's not
> > configured
> > > > for
> > > > > > ZK. Did I understand your question correctly?
> > > > > >
> > > > > > 3. Yea, good idea. Whil

Jenkins build is still unstable: Kafka » Kafka Branch Builder » trunk #1345

[Apache Jenkins Server]

See 

Re: [DISCUSS] KIP-866 ZooKeeper to KRaft Migration

[Jun Rao]

Hi, David,

Thanks for the reply.

20/21. Thanks for the explanation. The suggestion sounds good. Could you
include that in the doc?

40. metadata.migration.enable: We may do future metadata migrations within
KRaft. Could we make the name more specific to ZK migration like
zookeeper.metadata.migration.enable?

Thanks,

Jun

On Tue, Nov 8, 2022 at 9:47 AM David Arthur  wrote:

> Ah, sorry about that, you're right. Since we won't support ZK
> migrations in combined mode, is this issue avoided?
>
> Essentially, we would only set ZkMigrationReady in ApiVersionsResponse if
>
> * process.roles=controller
> * kafka.metadata.migration.enabled=true
>
> Also, the following would be an invalid config that should prevent startup:
>
> * process.roles=broker,controller
> * kafka.metadata.migration.enabled=true
>
> Does this seem reasonable?
>
> Thanks!
> David
>
> On Tue, Nov 8, 2022 at 11:12 AM Jun Rao  wrote:
> >
> > Hi, David,
> >
> > I am not sure that we are fully settled on the following.
> >
> > 20/21. Since separate listeners are optional, it seems that the broker
> > can't distinguish between ApiVersion requests coming from the client or
> > other brokers. This means the clients will get ZkMigrationReady in the
> > ApiVersion response, which is weird.
> >
> > Thanks,
> >
> > Jun
> >
> > On Tue, Nov 8, 2022 at 7:18 AM David Arthur  wrote:
> >
> > > Thanks for the discussion everyone, I'm going to move ahead with the
> > > vote for this KIP.
> > >
> > > -David
> > >
> > > On Thu, Nov 3, 2022 at 1:20 PM Jun Rao 
> wrote:
> > > >
> > > > Hi, David,
> > > >
> > > > Thanks for the reply.
> > > >
> > > > 20/21 Yes, but separate listeners are optional. It's possible for the
> > > nodes
> > > > to use a single port for both client and server side communications.
> > > >
> > > > Thanks,
> > > >
> > > > Jun
> > > >
> > > > On Thu, Nov 3, 2022 at 9:59 AM David Arthur
> > > >  wrote:
> > > >
> > > > > 20/21, in combined mode we still have a separate listener for the
> > > > > controller APIs, e.g.,
> > > > >
> > > > > listeners=PLAINTEXT://:9092,CONTROLLER://:9093
> > > > >
> > > > > inter.broker.listener.name=PLAINTEXT
> > > > >
> > > > > controller.listener.names=CONTROLLER
> > > > >
> > > > > advertised.listeners=PLAINTEXT://localhost:9092
> > > > >
> > > > >
> > > > >
> > > > > Clients still talk to the broker through the advertised listener,
> and
> > > only
> > > > > brokers and other controllers will communicate over the controller
> > > > > listener.
> > > > >
> > > > > 40. Sounds good, I updated the KIP
> > > > >
> > > > > Thanks!
> > > > > David
> > > > >
> > > > >
> > > > >
> > > > > On Thu, Nov 3, 2022 at 12:14 PM Jun Rao 
> > > wrote:
> > > > >
> > > > > > Hi, David,
> > > > > >
> > > > > > Thanks for the reply.
> > > > > >
> > > > > > 20/21. When KRaft runs in the combined mode, does a controller
> know
> > > > > whether
> > > > > > an ApiRequest is from a client or another broker?
> > > > > >
> > > > > > 40. Adding a "None" state sounds reasonable.
> > > > > >
> > > > > > Thanks,
> > > > > >
> > > > > > Jun
> > > > > >
> > > > > > On Thu, Nov 3, 2022 at 8:39 AM David Arthur
> > > > > >  wrote:
> > > > > >
> > > > > > > Jun,
> > > > > > >
> > > > > > > 20/21 If we use a tagged field, then I don't think clients
> need to
> > > be
> > > > > > > concerned with it, right?. In ApiVersionsResponse sent by
> brokers
> > > to
> > > > > > > clients, this field would be omitted. Clients can't connect
> > > directly to
> > > > > > the
> > > > > > > KRaft controller nodes. Also, we have a precedent of sending
> > > controller
> > > > > > > node state between controllers through ApiVersions
> > > ("metadata.version"
> > > > > > > feature), so I think it fits well.
> > > > > > >
> > > > > > > 40. For new KRaft clusters, we could add another state to
> indicate
> > > it
> > > > > was
> > > > > > > not migrated from ZK, but created as a KRaft cluster. Maybe
> > > > > > > "kafka.controller:type=KafkaController,name=MigrationState" =>
> > > "None" ?
> > > > > > We
> > > > > > > could also omit that metric for unmigrated clusters, but I'm
> not a
> > > fan
> > > > > of
> > > > > > > using the absence of a value to signal something.
> > > > > > >
> > > > > > > -
> > > > > > >
> > > > > > > Akhilesh, thanks for reviewing the KIP!
> > > > > > >
> > > > > > > 1. MigrationState and MetadataType are mostly the same on the
> > > > > controller,
> > > > > > > but we do have the "MigratingZkData" state that seems useful to
> > > report
> > > > > > as a
> > > > > > > metric. Aside from looking at logs, observing the controller in
> > > this
> > > > > > state
> > > > > > > is the only way to see how long its taking to copy data from
> ZK.
> > > > > > >
> > > > > > > "KRaftMode" instead of "MigrationFinalized" is similar to Jun's
> > > > > question
> > > > > > > about non-migrated clusters. I think it's useful to have a
> distinct
> > > > > > > MigrationState for clusters that have been migrated and those
> that
> > > were
> > > > > > >

Jenkins build is still unstable: Kafka » Kafka Branch Builder » trunk #1346

[Apache Jenkins Server]

See 

[jira] [Created] (KAFKA-14370) Properly close ImageWriter objects

[Colin McCabe (Jira)]

Colin McCabe created KAFKA-14370:


 Summary: Properly close ImageWriter objects
 Key: KAFKA-14370
 URL: https://issues.apache.org/jira/browse/KAFKA-14370
 Project: Kafka
  Issue Type: Bug
Reporter: Colin McCabe
Assignee: Colin McCabe






--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Re: [VOTE] KIP-866 ZooKeeper to KRaft Migration

[Colin McCabe]

Hi David,

Looks great. Some questions:

I agree with Jun that it would be good to rename metadata.migration.enable to 
something more zk-specific. Maybe zk.metadata.migration.enable ?

SImilarly, for MigrationRecord: can we rename this to ZkMigrationStateRecord? 
Then change MigrationState -> ZkMigrationState.

With ZkMigrationStateRecord, one thing to keep in mind here is that we will 
eventually compact all the metadata logs into a snapshot. That snapshot will 
then have to keep alive the memory of the old migration. So it is not really a 
matter of replaying the old metadata logs (probably) but a matter of checking 
to see what the ZkMigrationState is, which I suppose could be 
Optional. If it's not Optional.empty, we already migrated / 
are migrating.

For the /migration ZNode, is "last_update_time_ms" necessary? I thought ZK 
already tracked this information in the mzxid of the znode?

It is true that technically it is only needed in UMR, but I would still suggest 
including KRaftControllerId in LeaderAndIsrRequest because it will make 
debugging much easier.

I would suggest not implementing the topic deletion state machine, but just 
deleting topics eagerly when in migration mode. We can implement this behavior 
change by keying off of whether KRaftControllerId is present in 
LeaderAndIsrRequest. On broker startup, we'll be sent a full 
LeaderAndIsrRequest and can delete stray partitions whose IDs are not as 
expected (again, this behavior change would only be for migration mode)

For existing KRaft controllers, will 
kafka.controller:type=KafkaController,name=MigrationState show up as 4 
(MigrationFinalized)? I assume this is true, but it would be good to spell it 
out. Sorry if this is answered somewhere else.

As you point out, the ZK brokers being upgraded will need to contact the KRaft 
quorum in order to forward requests to there, once we are in migration mode. 
This raises a question: rather than changing the broker registration, can we 
have those brokers send an RPC to the kraft controller quorum instead? This 
would serve to confirm that they can reach the quorum. Then the quorum could 
wait for all of the brokers to check in this way before starting the migration 
(It would know all the brokers by looking at /brokers)

best,
Colin

On Tue, Nov 8, 2022, at 07:23, David Arthur wrote:
> Hello everyone, I'd like to start the vote on KIP-866.
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-866+ZooKeeper+to+KRaft+Migration
>
> Thanks!
> David Arthur

[jira] [Reopened] (KAFKA-13891) sync group failed with rebalanceInProgress error cause rebalance many rounds in coopeartive

[A. Sophie Blee-Goldman (Jira)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-13891?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

A. Sophie Blee-Goldman reopened KAFKA-13891:


Reopening – original fix was reverted, we should instead fix this assignor-side 
by making it smarter about partition ownership across generations. Basically, 
it should take as the previous owner whichever consumer has the highest 
generation and claims it among their owned partitions

 

[~showuon] I probably won't be able to get to this within the next few days so 
if you're interested in picking up this fix go ahead and I'll find time to 
review – otherwise I will try to get to it in time for the 3.4 release

> sync group failed with rebalanceInProgress error cause rebalance many rounds 
> in coopeartive
> ---
>
> Key: KAFKA-13891
> URL: https://issues.apache.org/jira/browse/KAFKA-13891
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 3.0.0
>Reporter: Shawn Wang
>Priority: Major
> Fix For: 3.3.0, 3.2.4
>
>
> This issue was first found in 
> [KAFKA-13419|https://issues.apache.org/jira/browse/KAFKA-13419]
> But the previous PR forgot to reset generation when sync group failed with 
> rebalanceInProgress error. So the previous bug still exists and it may cause 
> consumer to rebalance many rounds before final stable.
> Here's the example ({*}bold is added{*}):
>  # consumer A joined and synced group successfully with generation 1 *( with 
> ownedPartition P1/P2 )*
>  # New rebalance started with generation 2, consumer A joined successfully, 
> but somehow, consumer A doesn't send out sync group immediately
>  # other consumer completed sync group successfully in generation 2, except 
> consumer A.
>  # After consumer A send out sync group, the new rebalance start, with 
> generation 3. So consumer A got REBALANCE_IN_PROGRESS error with sync group 
> response
>  # When receiving REBALANCE_IN_PROGRESS, we re-join the group, with 
> generation 3, with the assignment (ownedPartition) in generation 1.
>  # So, now, we have out-of-date ownedPartition sent, with unexpected results 
> happened
>  # *After the generation-3 rebalance, consumer A got P3/P4 partition. the 
> ownedPartition is ignored because of old generation.*
>  # *consumer A revoke P1/P2 and re-join to start a new round of rebalance*
>  # *if some other consumer C failed to syncGroup before consumer A's 
> joinGroup. the same issue will happens again and result in many rounds of 
> rebalance before stable*
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

[jira] [Resolved] (KAFKA-14363) Add new `group-coordinator` module

[David Jacot (Jira)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-14363?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

David Jacot resolved KAFKA-14363.
-
Resolution: Fixed

> Add new `group-coordinator` module
> --
>
> Key: KAFKA-14363
> URL: https://issues.apache.org/jira/browse/KAFKA-14363
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: David Jacot
>Assignee: David Jacot
>Priority: Major
>




--
This message was sent by Atlassian Jira
(v8.20.10#820010)

RE: Last sprint to finish line: Replace EasyMock/Powermock with Mockito

[Christo Lolov]

Hello!

This email summarises the current state of Kafka's Mockito migration.

The JIRA tickets used to track the progress are 
https://issues.apache.org/jira/browse/KAFKA-14132 
 and 
https://issues.apache.org/jira/browse/KAFKA-14133 
.

—

Breakdown of https://issues.apache.org/jira/browse/KAFKA-14133 

19/46 ~ 41% are merged
27/46 ~ 59% are in review

A list of pull requests awaiting a review from a committer:
https://github.com/apache/kafka/pull/12739 

https://github.com/apache/kafka/pull/12505 

https://github.com/apache/kafka/pull/12524 

https://github.com/apache/kafka/pull/12818 


—

Breakdown of https://issues.apache.org/jira/browse/KAFKA-14132 

7/17 ~ 41% are merged
6/17 ~ 35% are in review
4/17 ~ 24% are in progress

A list of pull requests awaiting a review from a committer:
https://github.com/apache/kafka/pull/12728 

https://github.com/apache/kafka/pull/12821 


—

A list of pull requests which have been merged since the last update:
https://github.com/apache/kafka/pull/12527 

https://github.com/apache/kafka/pull/12725 

https://github.com/apache/kafka/pull/12823 


A big thank you to Shekhar Prasad Rajak (who recently joined our effort), 
Matthew de Detrich, Dalibor Plavcic, and everyone who has provided reviews over 
the last month!

Best,
Christo