You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2023/05/26 17:47:03 UTC
[kafka] branch 3.5 updated: KAFKA-15007: Use the correct MetadataVersion in MetadataPropagator (#13732)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch 3.5
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.5 by this push:
new 67247beb364 KAFKA-15007: Use the correct MetadataVersion in MetadataPropagator (#13732)
67247beb364 is described below
commit 67247beb3644d7912c69077bdcd57fff9869758d
Author: Akhilesh C <ak...@users.noreply.github.com>
AuthorDate: Mon May 22 11:46:53 2023 -0700
KAFKA-15007: Use the correct MetadataVersion in MetadataPropagator (#13732)
Use the MetadataVersion from the MetadataImage passed to MetadataPropagator. The ensures the propagator
sends the right versions of UMR, LISR and StopReplica requests when the migration is in DUAL_WRITE mode.
Reviewers: David Arthur <mu...@gmail.com>
---
core/src/main/scala/kafka/migration/MigrationPropagator.scala | 8 +-------
.../org/apache/kafka/metadata/migration/KRaftMigrationDriver.java | 3 ---
.../org/apache/kafka/metadata/migration/LegacyPropagator.java | 3 ---
.../apache/kafka/metadata/migration/KRaftMigrationDriverTest.java | 6 ------
4 files changed, 1 insertion(+), 19 deletions(-)
diff --git a/core/src/main/scala/kafka/migration/MigrationPropagator.scala b/core/src/main/scala/kafka/migration/MigrationPropagator.scala
index e7bbd11edd9..c573991b84e 100644
--- a/core/src/main/scala/kafka/migration/MigrationPropagator.scala
+++ b/core/src/main/scala/kafka/migration/MigrationPropagator.scala
@@ -26,7 +26,6 @@ import org.apache.kafka.common.utils.Time
import org.apache.kafka.image.{ClusterImage, MetadataDelta, MetadataImage, TopicsImage}
import org.apache.kafka.metadata.PartitionRegistration
import org.apache.kafka.metadata.migration.LegacyPropagator
-import org.apache.kafka.server.common.MetadataVersion
import java.util
import scala.jdk.CollectionConverters._
@@ -57,7 +56,6 @@ class MigrationPropagator(
config: KafkaConfig
) extends LegacyPropagator {
@volatile private var _image = MetadataImage.EMPTY
- @volatile private var metadataVersion = MetadataVersion.IBP_3_4_IV0
val stateChangeLogger = new StateChangeLogger(nodeId, inControllerContext = true, None)
val channelManager = new ControllerChannelManager(
() => _image.highestOffsetAndEpoch().epoch(),
@@ -70,7 +68,7 @@ class MigrationPropagator(
val requestBatch = new MigrationPropagatorBatch(
config,
metadataProvider,
- () => metadataVersion,
+ () => _image.features().metadataVersion(),
channelManager,
stateChangeLogger
)
@@ -248,8 +246,4 @@ class MigrationPropagator(
override def clear(): Unit = {
requestBatch.clear()
}
-
- override def setMetadataVersion(newMetadataVersion: MetadataVersion): Unit = {
- metadataVersion = newMetadataVersion
- }
}
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
index caa030abbce..213d86ca4e8 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java
@@ -629,9 +629,6 @@ public class KRaftMigrationDriver implements MetadataPublisher {
completionHandler.accept(null);
return;
}
- if (delta.featuresDelta() != null) {
- propagator.setMetadataVersion(image.features().metadataVersion());
- }
if (image.highestOffsetAndEpoch().compareTo(migrationLeadershipState.offsetAndEpoch()) < 0) {
log.info("Ignoring {} {} which contains metadata that has already been written to ZK.", metadataType, provenance);
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java
index 6fe707099a3..c2bd4e6f3d0 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/LegacyPropagator.java
@@ -18,7 +18,6 @@ package org.apache.kafka.metadata.migration;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
-import org.apache.kafka.server.common.MetadataVersion;
public interface LegacyPropagator {
@@ -35,6 +34,4 @@ public interface LegacyPropagator {
void sendRPCsToBrokersFromMetadataImage(MetadataImage image, int zkControllerEpoch);
void clear();
-
- void setMetadataVersion(MetadataVersion metadataVersion);
}
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java b/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
index 0f84a068b81..afbc952c288 100644
--- a/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
+++ b/metadata/src/test/java/org/apache/kafka/metadata/migration/KRaftMigrationDriverTest.java
@@ -46,7 +46,6 @@ import org.apache.kafka.metadata.RecordTestUtils;
import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.raft.OffsetAndEpoch;
import org.apache.kafka.server.common.ApiMessageAndVersion;
-import org.apache.kafka.server.common.MetadataVersion;
import org.apache.kafka.server.fault.MockFaultHandler;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Assertions;
@@ -163,11 +162,6 @@ public class KRaftMigrationDriverTest {
public void clear() {
}
-
- @Override
- public void setMetadataVersion(MetadataVersion metadataVersion) {
-
- }
}
RegisterBrokerRecord zkBrokerRecord(int id) {