KAFKA-18076: Remove isZkMigrationTest and related code

core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala

@@ -70,9 +70,6 @@ abstract class IntegrationTestHarness extends KafkaServerTestHarness {
       trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount)
     configureListeners(cfgs)
     modifyConfigs(cfgs)
-    if (isZkMigrationTest()) {
-      cfgs.foreach(_.setProperty(KRaftConfigs.MIGRATION_ENABLED_CONFIG, "true"))
-    }
     if (isShareGroupTest()) {
       cfgs.foreach(_.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,share"))
       cfgs.foreach(_.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true"))

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

@@ -197,10 +197,6 @@ abstract class QuorumTestHarness extends Logging {
     TestInfoUtils.isKRaft(testInfo)
   }
 
-  def isZkMigrationTest(): Boolean = {
-    TestInfoUtils.isZkMigrationTest(testInfo)
-  }
-
   def isShareGroupTest(): Boolean = {
     TestInfoUtils.isShareGroupTest(testInfo)
   }

core/src/test/scala/kafka/utils/TestInfoUtils.scala

@@ -45,14 +45,6 @@ object TestInfoUtils {
     }
   }
 
-  def isZkMigrationTest(testInfo: TestInfo): Boolean = {
-    if (!isKRaft(testInfo)) {
-      false
-    } else {
-      testInfo.getDisplayName.contains("quorum=zkMigration")
-    }
-  }
-
   final val TestWithParameterizedQuorumAndGroupProtocolNames = "{displayName}.quorum={0}.groupProtocol={1}"
 
   def isShareGroupTest(testInfo: TestInfo): Boolean = {

core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala

@@ -413,7 +413,7 @@ abstract class KafkaServerTestHarness extends QuorumTestHarness {
         time = brokerTime(config.brokerId),
         threadNamePrefix = None,
         startup = false,
-        enableZkApiForwarding = isZkMigrationTest() || (config.migrationEnabled && config.interBrokerProtocolVersion.isApiForwardingEnabled)
+        enableZkApiForwarding = config.migrationEnabled && config.interBrokerProtocolVersion.isApiForwardingEnabled
       )
     }
   }

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

@@ -18,7 +18,6 @@
 package kafka.server
 
 import java.util
-import java.util.Collections
 import kafka.network.SocketServer
 import kafka.utils.{Logging, TestUtils}
 import org.apache.kafka.common.Uuid
@@ -216,24 +215,6 @@ class DeleteTopicsRequestTest extends BaseRequestTest with Logging {
     }
   }
 
-  /*
-   * Only run this test against ZK clusters. KRaft doesn't have this behavior of returning NOT_CONTROLLER.
-   * Instead, the request is forwarded.
-   */
-  @ParameterizedTest
-  @ValueSource(strings = Array("zk", "zkMigration"))
-  def testNotController(quorum: String): Unit = {
-    val request = new DeleteTopicsRequest.Builder(
-        new DeleteTopicsRequestData()
-          .setTopicNames(Collections.singletonList("not-controller"))
-          .setTimeoutMs(1000)).build()
-    val response = sendDeleteTopicsRequest(request, notControllerSocketServer)
-
-    val expectedError = if (isZkMigrationTest()) Errors.NONE else Errors.NOT_CONTROLLER
-    val error = response.data.responses.find("not-controller").errorCode()
-    assertEquals(expectedError.code(),  error)
-  }
-
   private def validateTopicIsDeleted(topic: String): Unit = {
     val metadata = connectAndReceive[MetadataResponse](new MetadataRequest.Builder(
       List(topic).asJava, true).build).topicMetadata.asScala