Assert bridge opt-ins at request and storage boundaries

core/src/test/scala/integration/kafka/server/LiReassignmentCancellationGateTest.scala

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server
+
+import kafka.api.LeaderAndIsr
+import kafka.controller.{LeaderIsrAndControllerEpoch, ReplicaAssignment}
+import kafka.utils.TestUtils
+import kafka.zk.ZkVersion
+import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, NewPartitionReassignment}
+import org.apache.kafka.common.{TopicPartition, Uuid}
+import org.apache.kafka.common.errors.InvalidReplicaAssignmentException
+import org.apache.kafka.server.config.ServerConfigs
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue}
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+
+import java.util.{Collections, Optional, Properties}
+import java.util.concurrent.{ExecutionException, TimeUnit}
+
+class LiReassignmentCancellationGateTest extends QuorumTestHarness {
+  private var broker: KafkaServer = _
+
+  @AfterEach
+  override def tearDown(): Unit = {
+    if (broker != null) broker.shutdown()
+    super.tearDown()
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = Array(false, true))
+  def testCancellationGateAtControllerRequestBoundary(enabled: Boolean): Unit = {
+    val tp = new TopicPartition("cancel-gate", 0)
+    val topicId = Some(Uuid.randomUuid())
+    // Resume an existing move from [0,1] to [2], with only original replica 0 online.
+    // Seeding before startup avoids a race with a destination catching up in the fixture.
+    zkClient.createTopicAssignment(tp.topic, topicId, Map(tp -> Seq(0, 1)))
+    zkClient.setTopicAssignment(tp.topic, topicId,
+      Map(tp -> ReplicaAssignment(Seq(2, 0, 1), Seq(2), Seq(0, 1))))
+    zkClient.createTopicPartitionStatesRaw(
+      Map(tp -> LeaderIsrAndControllerEpoch(LeaderAndIsr(0, List(0)), 0)), ZkVersion.MatchAnyVersion)
+      .foreach(_.maybeThrow())
+    val props = TestUtils.createBrokerConfig(0, zkConnect)
+    props.put(ServerConfigs.CONTROLLED_SHUTDOWN_ENABLE_CONFIG, "false")
+    props.put(KafkaConfig.LiMinOriginalAliveReplicasProp, "2")
+    if (enabled) props.put(KafkaConfig.LiProtocolBridgeReassignmentCancellationSafetyEnableProp, "true")
+    broker = createBroker(KafkaConfig.fromProps(props)).asInstanceOf[KafkaServer]
+    assertEquals(0, TestUtils.waitUntilControllerElected(zkClient))
+    val adminProps = new Properties
+    adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+      TestUtils.bootstrapServers(Seq(broker), broker.config.interBrokerListenerName))
+    val admin = Admin.create(adminProps)
+    try {
+      val result = admin.alterPartitionReassignments(
+        Collections.singletonMap(tp, Optional.empty[NewPartitionReassignment]())).all()
+      if (enabled) {
+        val error = assertThrows(classOf[ExecutionException], () => result.get(10, TimeUnit.SECONDS))
+        assertTrue(error.getCause.isInstanceOf[InvalidReplicaAssignmentException])
+      } else result.get(10, TimeUnit.SECONDS)
+      val assignment = zkClient.getFullReplicaAssignmentForTopics(Set(tp.topic))(tp)
+      assertEquals(if (enabled) Seq(2) else Seq(0, 1), assignment.targetReplicas)
+    } finally admin.close()
+  }
+}

core/src/test/scala/unit/kafka/cluster/PartitionTest.scala

@@ -2154,8 +2154,9 @@ class PartitionTest extends AbstractPartitionTest {
     assertEquals(log.logEndOffset, partition.localLogOrException.highWatermark)
   }
 
-  @Test
-  def testLeaderTransferSelectsLowestInSyncFollower(): Unit = {
+  @ParameterizedTest
+  @ValueSource(booleans = Array(false, true))
+  def testLeaderTransferSelectsLowestInSyncFollower(enabled: Boolean): Unit = {
     configRepository.setTopicConfig(topicPartition.topic, TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "4")
     val log = logManager.getOrCreateLog(topicPartition, topicId = None)
     seedLogData(log, numRecords = 10, leaderEpoch = 4)
@@ -2177,7 +2178,7 @@ class PartitionTest extends AbstractPartitionTest {
       metadataCache,
       logManager,
       alterPartitionManager,
-      leaderTransferEnabled = true,
+      leaderTransferEnabled = enabled,
       leaderTransferManager = leaderTransferManager)
     transferPartition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, None)
     assertTrue(transferPartition.makeLeader(
@@ -2196,9 +2197,11 @@ class PartitionTest extends AbstractPartitionTest {
     fetchFollower(transferPartition, replicaId = lowestInSyncFollower, fetchOffset = log.logEndOffset)
     time.sleep(transferPartition.replicaLagTimeMaxMs + 1)
     transferPartition.maybeTransferToNewLeader()
-
-    verify(leaderTransferManager).submit(topicPartition, lowestInSyncFollower)
+    if (enabled) verify(leaderTransferManager).submit(topicPartition, lowestInSyncFollower)
     verifyNoMoreInteractions(leaderTransferManager)
+
+    transferPartition.maybeShrinkIsr()
+    assertEquals(if (enabled) 0 else 1, alterPartitionManager.isrUpdates.size)
   }
 
   @Test

core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala

@@ -317,20 +317,25 @@ class DynamicBrokerConfigTest {
       KafkaConfig.LiProtocolBridgeReassignmentCancellationSafetyEnableProp,
       KafkaConfig.LiProtocolBridgeProduceRequestInstrumentationEnableProp
     )
-    val allBridgeFlags = dynamicBridgeFlags :+ KafkaConfig.LiProtocolBridgeLeaderTransferEnableProp
+    val allBridgeFlags = KafkaConfig.LiProtocolBridgeEnableProps
     allBridgeFlags.foreach(flag => assertFalse(config.getBoolean(flag), s"$flag should be disabled by default"))
 
     val dynamicProps = new Properties
     dynamicBridgeFlags.foreach(dynamicProps.put(_, "true"))
     config.dynamicConfig.validate(dynamicProps, perBrokerConfig = false)
     config.dynamicConfig.updateDefaultConfig(dynamicProps)
     dynamicBridgeFlags.foreach(flag => assertTrue(config.getBoolean(flag), s"$flag should be enabled dynamically"))
-    assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(dynamicProps, perBrokerConfig = true))
-
-    val startupOnlyProps = new Properties
-    startupOnlyProps.put(KafkaConfig.LiProtocolBridgeLeaderTransferEnableProp, "true")
-    assertThrows(classOf[ConfigException],
-      () => config.dynamicConfig.validate(startupOnlyProps, perBrokerConfig = false))
+    // Test each flag alone so one rejected key cannot hide an allowed override.
+    allBridgeFlags.foreach { flag =>
+      val oneFlag = new Properties
+      oneFlag.put(flag, "true")
+      assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(oneFlag, perBrokerConfig = true),
+        s"$flag must reject a per-broker override")
+      if (!dynamicBridgeFlags.contains(flag)) {
+        assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(oneFlag, perBrokerConfig = false),
+          s"$flag must require a restart")
+      }
+    }
   }
 
   @Test

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -229,6 +229,90 @@ class KafkaApisTest extends Logging {
       clientMetricsManager = clientMetricsManagerOpt)
   }
 
+  @ParameterizedTest
+  @CsvSource(Array("false,false,true", "false,true,true", "true,false,true", "true,true,true",
+    "false,true,false", "true,true,false"))
+  def testRecommendedElectionRequestRequiresCompatibilityGate(enabled: Boolean, authorized: Boolean,
+                                                               recommended: Boolean): Unit = {
+    val authorizer = mock(classOf[Authorizer])
+    authorizeResource(authorizer, AclOperation.ALTER, ResourceType.CLUSTER, Resource.CLUSTER_NAME,
+      if (authorized) AuthorizationResult.ALLOWED else AuthorizationResult.DENIED)
+    val tp = new TopicPartition("recommended", 0)
+    val electionType = if (recommended) ElectionType.RECOMMENDED else ElectionType.PREFERRED
+    val leaders = if (recommended) Map(tp -> 1) else Map.empty[TopicPartition, Int]
+    val builder = if (recommended) new ElectLeadersRequest.Builder(10L,
+      Collections.singletonMap(tp, Integer.valueOf(1)), 30000)
+    else new ElectLeadersRequest.Builder(electionType, Collections.singleton(tp), 30000)
+    val request = buildRequest(builder.build(2.toShort))
+    doAnswer { invocation =>
+      invocation.getArgument[Map[TopicPartition, ApiError] => Unit](4)(Map(tp -> ApiError.NONE))
+      null
+    }.when(replicaManager).electLeaders(ArgumentMatchers.eq(controller), ArgumentMatchers.eq(Set(tp)),
+      ArgumentMatchers.eq(leaders), ArgumentMatchers.eq(electionType), any(), ArgumentMatchers.eq(30000))
+    kafkaApis = createKafkaApis(authorizer = Some(authorizer), overrideProperties = Map(
+      KafkaConfig.LiProtocolBridgeRecommendedElectionEnableProp -> enabled.toString))
+    kafkaApis.handleElectLeaders(request)
+    val response = verifyNoThrottling[ElectLeadersResponse](request)
+    val expectedError = if (recommended && !enabled) Errors.INVALID_REQUEST
+      else if (!authorized) Errors.CLUSTER_AUTHORIZATION_FAILED else Errors.NONE
+    assertEquals(expectedError.code, response.data.errorCode)
+    verify(replicaManager, times(if ((!recommended || enabled) && authorized) 1 else 0))
+      .electLeaders(any(), any(), any(), any(), any(), anyInt())
+  }
+
+  @ParameterizedTest
+  @CsvSource(Array("false,false", "false,true", "true,false", "true,true"))
+  def testMetadataExclusionRequiresRequestAndCompatibilityGate(enabled: Boolean, requested: Boolean): Unit = {
+    val topic = "metadata-exclusion"
+    addTopicToMetadataCache(topic, numPartitions = 2, numBrokers = 1)
+    val data = new MetadataRequestData()
+      .setTopics(Collections.singletonList(new MetadataRequestData.MetadataRequestTopic().setName(topic)))
+      .setAllowAutoTopicCreation(false)
+      .setExcludePartitions(requested)
+    val request = buildRequest(new MetadataRequest(data, 12.toShort))
+    kafkaApis = createKafkaApis(overrideProperties = Map(
+      KafkaConfig.LiProtocolBridgeExcludePartitionsEnableProp -> enabled.toString))
+    kafkaApis.handleTopicMetadataRequest(request)
+    val response = verifyNoThrottling[MetadataResponse](request)
+    assertEquals(1, response.topicMetadata.size)
+    val metadata = response.topicMetadata.iterator.next()
+    assertEquals(Errors.NONE, metadata.error)
+    assertEquals(topic, metadata.topic)
+    assertEquals(if (enabled && requested) 0 else 2, metadata.partitionMetadata.size)
+  }
+
+  @ParameterizedTest
+  @CsvSource(Array("false,-104,7,false", "true,-104,6,false", "true,-104,7,false",
+    "true,-104,7,true", "false,-4,8,true", "true,-4,8,true"))
+  def testFollowerRecoveryRequestAndErrorGates(enabled: Boolean, timestamp: Long,
+                                               version: Short, moved: Boolean): Unit = {
+    val tp = new TopicPartition("recovery-gate", 0)
+    val fetch = when(replicaManager.fetchOffsetForTimestamp(ArgumentMatchers.eq(tp), ArgumentMatchers.eq(timestamp),
+      any[Option[IsolationLevel]](), any[Optional[Integer]](), anyBoolean()))
+    if (moved) fetch.thenThrow(Errors.OFFSET_MOVED_TO_TIERED_STORAGE.exception)
+    else fetch.thenReturn(Some(new TimestampAndOffset(0L, 12L, Optional.of[Integer](1))))
+    val topic = new ListOffsetsTopic().setName(tp.topic).setPartitions(Collections.singletonList(
+      new ListOffsetsPartition().setPartitionIndex(tp.partition).setTimestamp(timestamp)))
+    val request = buildRequest(ListOffsetsRequest.Builder.forConsumer(true, IsolationLevel.READ_UNCOMMITTED)
+      .setTargetTimes(Collections.singletonList(topic)).build(version))
+    kafkaApis = createKafkaApis(overrideProperties = Map(
+      KafkaConfig.LiProtocolBridgeFollowerRecoveryEnableProp -> enabled.toString))
+    kafkaApis.handleListOffsetRequest(request)
+    val response = verifyNoThrottling[ListOffsetsResponse](request)
+    assertEquals(1, response.topics.size)
+    val partition = response.topics.get(0).partitions.get(0)
+    val legacy = timestamp == ListOffsetsRequest.LI_EARLIEST_LOCAL_TIMESTAMP
+    val admitted = !legacy || (enabled && version >= 7)
+    val expectedError = if (!admitted) Errors.UNSUPPORTED_VERSION.code
+      else if (moved && legacy) 1107.toShort
+      else if (moved) Errors.OFFSET_MOVED_TO_TIERED_STORAGE.code
+      else Errors.NONE.code
+    assertEquals(expectedError, partition.errorCode)
+    assertEquals(if (admitted && !moved) 12L else ListOffsetsResponse.UNKNOWN_OFFSET, partition.offset)
+    verify(replicaManager, times(if (admitted) 1 else 0)).fetchOffsetForTimestamp(
+      ArgumentMatchers.eq(tp), ArgumentMatchers.eq(timestamp), any(), any(), anyBoolean())
+  }
+
   @Test
   def testDescribeConfigsWithAuthorizer(): Unit = {
     val authorizer: Authorizer = mock(classOf[Authorizer])

core/src/test/scala/unit/kafka/server/LiProtocolBridgeConfigTest.scala

@@ -50,6 +50,16 @@ class LiProtocolBridgeConfigTest {
     assertEquals(true, enabled.extractLogConfigMap.get(LogConfig.LI_LOG_TRUNCATION_METRICS_CONFIG))
   }
 
+  @Test
+  def testNonzeroMinimumLogRollRequiresCompatibilityGate(): Unit = {
+    val props = new Properties
+    props.put(ZkConfigs.ZK_CONNECT_CONFIG, "localhost:2181")
+    props.put(KafkaConfig.LiMinLogRollTimeMillisProp, "60000")
+    assertEquals(0L, KafkaConfig.fromProps(props).extractLogConfigMap.get(LogConfig.LI_MIN_SEGMENT_MS_CONFIG))
+    props.put(KafkaConfig.LiProtocolBridgeMinimumLogRollEnableProp, "true")
+    assertEquals(60000L, KafkaConfig.fromProps(props).extractLogConfigMap.get(LogConfig.LI_MIN_SEGMENT_MS_CONFIG))
+  }
+
   @Test
   def testBridgeModeRejectsMetadataThatLeaderAndIsrV2CannotRepresent(): Unit = {
     val props = new Properties