KAFKA-17783: Adding listeners to remove share partition on partition changes

[apoorvmittal10]

The PR adds listeners to Partition class which might invoke the clean up of share partitions on Partition changes.

I considered introducing new Listener class which defines methods per SharePartitionKey but avoided as that would have required wiring up at the calls at different places in ReplicaManager. Additionally replica manager needed handling to store the new listener, etc. I found the current PR approach better as SharePartition is abstracted Shared view of Partition hence attaching the change events on Partition is helpful.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[AndrewJSchofield]

This looks like a promising approach. It's in an area of code that I really don't know, so I'm not qualified to a proper review. I do have a couple of comments:

• Can the partition become a leader after it has previously become a follower? I just wonder whether the state machine is more complicated than this PR implies?
• The listener is called under a lock in the Partition. The listener implementation needs to be very careful not to do anything brave under that lock. I expect there is potential for a deadlock here.

[apoorvmittal10]

Thanks @AndrewJSchofield for review and good points.

Can the partition become a leader after it has previously become a follower? I just wonder whether the state machine is more complicated than this PR implies?

Yes, that can always happen. Once it does then next share fetch request should load the share partition. The scenario is not very different than partition being shuffled across brokers. The broker who is the leader of the partition should be able to load share partition.

The listener is called under a lock in the Partition. The listener implementation needs to be very careful not to do anything brave under that lock. I expect there is potential for a deadlock here.

The lock is for each Partition, but I do get the point. I can make the call outside lock as well, I ll wait for @junrao's comments as he can help.

[mumrah]

@apoorvmittal10 what is the purpose of this listener? Is is so that SPM can clean up its in-memory state when it is no longer the leader for a partition?

If that's the main use case, I think we should consider tying into the metadata system directly rather than coupling ourselves to Partition.

@dajac, since its looks like you added it, maybe you can comment on the intended usage of PartitionListener?

[apoorvmittal10]

@apoorvmittal10 what is the purpose of this listener? Is is so that SPM can clean up its in-memory state when it is no longer the leader for a partition?

Yes, that's the purpose.

If that's the main use case, I think we should consider tying into the metadata system directly rather than coupling ourselves to Partition.

So as far as I see the code there exists 1 approach taken by coordinators: in replica manager where we have a method becomeLeaderOrFollower where we pass onLeadershipChange method. The onLeadershipChange method triggers groupCoordinator and txnCordinators, onElection and onResignation methods. The 2) approach is by invoking the listeners on partition change.

I am not sure if we do get leadership change notifications from metadata system. Not sure if we could have some kind of leaderEpochChange listener to see the partition epoch bump and do some actions. But I found the listeners approach much sensible.

[mumrah]

Thanks for the details, @apoorvmittal10. As we discussed offline, those methods in ReplicaManager are driven by the metadata system (BrokerMetadataPublisher). In the end, if we're just acting on one partition at a time, it shouldn't matter if we use the partition listener or the metadata publisher.

I am not sure if we do get leadership change notifications from metadata system

Partition leadership, ISR, etc are all controller-managed metadata which flow through the metadata system. We can reliably learn if a partition became a leader or if it became a follower through MetadataPublisher (this is how ReplicaManager and the coordinators work).

--

Thinking a bit more about this, I'd like to use a metadata publisher rather than modifying the PartitionListener. It seems that PartitionListener is more geared towards the log state rather than the partition metadata. In fact, the only production usage of this listener I can find is kafka.coordinator.group.ListenerAdapter#onHighWatermarkUpdated.

Since we are interested in subscribing to changes in the partition's leadership state, I think it is better to directly subscribe to the metadata system. This also has the benefit of eliminating the (potentially numerous) SharePartitionListener objects.

[mumrah]

INFO is probably too noisy for this as we expect it any time the topic leadership changes.

[apoorvmittal10]

Sure, can make it debug as well.

[apoorvmittal10]

Thanks @mumrah, works well with me. I was more thinking to tie it up with Partition listener initially but can work either way. @dajac @junrao What do you think?

[junrao]

@apoorvmittal10 : Thanks for the PR. Left a comment.

Regarding whether to use a listener at the Partition level or from BrokerMetadataPublisher, it's probably better to use a Partition level listener. It's true that the source of truth of the metadata is BrokerMetadataPublisher. However, the share partition requests only interact with metadata at the Partition level. So, it's more natural to take an action to share partitions after the metadata has been reflected at the Partition level. Otherwise, what could happen is that we remove a partition after the metadata changes in BrokerMetadataPublisher, but not in Partition yet. When we try to create the new SharePartition, it's still created with the outdated leaderEpoch.

[junrao]

If a broker switches from a follower to a follower, there is no need to remove the share partition. It's only when a broker is no longer a leader, a share partition should be removed. So we could add a more specific listener.

[apoorvmittal10]

Do you think either onLeaderToFollower or onBecomeFollower could be the better names?

[apoorvmittal10]

Done.

[apoorvmittal10]

@apoorvmittal10 : Thanks for the PR. Left a comment.

Regarding whether to use a listener at the Partition level or from BrokerMetadataPublisher, it's probably better to use a Partition level listener. It's true that the source of truth of the metadata is BrokerMetadataPublisher. However, the share partition requests only interact with metadata at the Partition level. So, it's more natural to take an action to share partitions after the metadata has been reflected at the Partition level. Otherwise, what could happen is that we remove a partition after the metadata changes in BrokerMetadataPublisher, but not in Partition yet. When we try to create the new SharePartition, it's still created with the outdated leaderEpoch.

Thanks a lot for confirming. Then I will go ahead with this approach and ll add tests to the PR.
cc: @mumrah @dajac

[apoorvmittal10]

@AndrewJSchofield @mumrah @junrao I have updated the code with tests. Also removed listener call from locked markFollower method, @AndrewJSchofield mentioned, rather moved to replicaManager itself. Please review.