KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remove shim #10497

hachikuji · 2021-04-07T03:19:14Z

This patch removes the temporary shim layer we added to bridge the interface differences between MetaLogManager and RaftClient.

Reverse dependency between :raft and :metadata modules.
Consolidate handleResign and handleNewLeader APIs into single handleLeaderChange API
Move MetadataRecordSerde into :metadata
Update listeners to use BatchReader which takes disk reads out of the Raft IO thread
Delete MetaLogRaftShim, MetaLogManager, and MetaLogListener

Committer Checklist (excluded from commit message)

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

…e shim

ijuma · 2021-04-07T15:15:42Z

cc @jsancio

jsancio

Thanks for the PR. Partial review. Wanted to provide feedback as soon as possible.

core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala

jsancio · 2021-04-07T17:49:53Z

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

                } else {
-                    offset = logManager.scheduleWrite(controllerEpoch, result.records());
+                    offset = raftClient.scheduleAppend(controllerEpoch, result.records());


Should we file an Jira, if one doesn't already exists, to handle the case of when this is null?

We have https://issues.apache.org/jira/browse/KAFKA-12158 for this issue.

@hachikuji I think for now we should at least check the return value with offset = Object.requireNonNull(...).

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

jsancio · 2021-04-27T22:15:13Z

core/src/main/scala/kafka/raft/RaftManager.scala

@@ -126,10 +130,10 @@ class KafkaRaftManager[T](
  private val dataDir = createDataDir()
  private val metadataLog = buildMetadataLog()
  private val netChannel = buildNetworkChannel()
-  private val raftClient = buildRaftClient()
-  private val raftIoThread = new RaftIoThread(raftClient, threadNamePrefix)
+  val client: KafkaRaftClient[T] = buildRaftClient()


Did you mean to override the return type from RaftClient[T] to KafkaRaftClient[T]?

jsancio · 2021-04-27T22:24:05Z

core/src/main/scala/kafka/raft/RaftManager.scala


-  def kafkaRaftClient: KafkaRaftClient[T] = raftClient
+  def kafkaRaftClient: KafkaRaftClient[T] = client


I think that since you added a new method client: RaftClient[T] to RaftManager[T] and KafkaRaftManager overrides it to client: KafkaRaftClient[T] we should be able to remove this KafkaRaftManager[T] only public method.

jsancio · 2021-04-27T22:44:05Z

core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala

    override def run(): Unit = {
+      try {
+        apply(reader.next())


I think this is technically correct based on the current implementation of KafkaRaftClient but is there a reason why the listener is only reading one batch? Also, should it assume that the reader contains at least one batch?

When handleCommit is fired because of an appendAsLeader the reader is guarantee to have one batch. When reading from the ReplicatedLog the batch reader may have more than one batch.

Should this method instead do the following?

override def run(): Unit = { try { while(reader.hasNext()) { apply(reader.next()) } } finally { reader.close() } }

Created https://issues.apache.org/jira/browse/KAFKA-12837 for this suggestion.

jsancio · 2021-04-27T22:52:55Z

core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala

+  private def applyBatch(
+    records: List[ApiMessageAndVersion]
+  ): Unit = {
+    val baseOffset = lastMetadataOffset + 1


Hmm. This is minor but does this mean that a baseOffset of 0 is not possible since lastMetadataOffset is initialized to 0? Is this also true for a "regular" Kafka topic partition? Or is this just an side effect of how this test gets constructed.

jsancio · 2021-04-27T22:57:10Z

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -200,7 +202,7 @@ public Builder setMetrics(ControllerMetrics controllerMetrics) {

        @SuppressWarnings("unchecked")
        public QuorumController build() throws Exception {
-            if (logManager == null) {
+            if (raftClient == null) {
                throw new RuntimeException("You must set a metadata log manager.");


Let's change the message for this exception. Maybe "Raft client was not set for the quorum controller"

Resolved in #10705

jsancio · 2021-04-27T23:59:58Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+            if (leaderAndEpoch.equals(lastFiredLeaderChange)) {
+                return false;
+            } else if (leaderAndEpoch.epoch > lastFiredLeaderChange.epoch) {
+                return true;


I see. We want to fire this event even if the leader is Optional.empty() because we use this event to propagate lost of leadership.

jsancio · 2021-04-28T00:01:46Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+            } else if (leaderAndEpoch.epoch > lastFiredLeaderChange.epoch) {
+                return true;
+            } else {
+                return leaderAndEpoch.leaderId.isPresent() && !lastFiredLeaderChange.leaderId.isPresent();


This works because there is an invariant that leaderAndEpoch.epoch >= lastFiredLeaderChange.epoch is always true, right? Should we document this above this line?

Thinking about it some more, wouldn't this always be true since we know that:

leaderAndEpoch.epoch == lastFiredLeaderChange.epoch

!leaderAndEpoch.equals(lastFiredLeaderChange)

If you agree, I think that we can change this method to just:

return !leaderAndEpoch.equals(lastFiredLeaderChange);

jsancio · 2021-04-28T00:09:23Z

raft/src/main/java/org/apache/kafka/raft/LeaderAndEpoch.java

@@ -28,6 +28,10 @@ public LeaderAndEpoch(OptionalInt leaderId, int epoch) {
        this.epoch = epoch;
    }

+    public boolean isLeader(int nodeId) {
+        return leaderId.isPresent() && leaderId.getAsInt() == nodeId;


Minor but how about return leaderId.equals(OptionalInt.of(nodeId));

jsancio · 2021-04-28T00:10:47Z

raft/src/main/java/org/apache/kafka/raft/RaftClient.java

-         * @param epoch the epoch that the leader is resigning from
-         */
-        default void handleResign(int epoch) {}
+        default void beginShutdown() {}


Let's document this method.

jsancio · 2021-04-28T00:17:05Z

shell/src/main/java/org/apache/kafka/shell/MetadataNodeManager.java

-
-        @Override
-        public void handleNewLeader(MetaLogLeader leader) {
+        public void handleLeaderChange(LeaderAndEpoch leader) {
            appendEvent("handleNewLeader", () -> {


This is probably outside the scope of this PR but it looks like queue is never read.

jsancio · 2021-05-15T19:56:22Z

This PR was moved to #10705. @ijuma @cmccabe feel free to close this PR.

Jason Gustafson added 2 commits April 6, 2021 20:18

KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remov…

bf5ed4b

…e shim

Merge remote-tracking branch 'upstream/trunk' into KAFKA-12342

49934ef

hachikuji added the kraft label Apr 7, 2021

jsancio reviewed Apr 7, 2021

View reviewed changes

Move reader iteration to event handler

2340d7f

hachikuji force-pushed the KAFKA-12342 branch from 6ab1120 to 2340d7f Compare April 7, 2021 19:04

jsancio mentioned this pull request Apr 9, 2021

KAFKA-12154: Snapshot Loading API #10085

Merged

3 tasks

Merge remote-tracking branch 'upstream/trunk' into KAFKA-12342

ac1bb30

jsancio reviewed Apr 28, 2021

View reviewed changes

ijuma closed this May 24, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remove shim #10497

KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remove shim #10497

hachikuji commented Apr 7, 2021

ijuma commented Apr 7, 2021

jsancio left a comment

jsancio Apr 7, 2021

jsancio Apr 27, 2021

jsancio Apr 27, 2021

jsancio Apr 27, 2021

jsancio Apr 27, 2021

jsancio May 21, 2021

jsancio Apr 27, 2021 •

edited

Loading

jsancio Apr 27, 2021

jsancio May 21, 2021

jsancio Apr 27, 2021

jsancio Apr 28, 2021

jsancio Apr 28, 2021

jsancio Apr 28, 2021

jsancio Apr 28, 2021

jsancio commented May 15, 2021


		def kafkaRaftClient: KafkaRaftClient[T] = raftClient
		def kafkaRaftClient: KafkaRaftClient[T] = client

KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remove shim #10497

KAFKA-12342; Merge RaftClient and MetaLogManager interfaces and remove shim #10497

Conversation

hachikuji commented Apr 7, 2021

Committer Checklist (excluded from commit message)

ijuma commented Apr 7, 2021

jsancio left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio Apr 27, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio commented May 15, 2021

jsancio Apr 27, 2021 •

edited

Loading