KAFKA-18529: ConsumerRebootstrapTest should run for async consumer (#18554)

Reviewers: Kirk True <[email protected]>, Chia-Ping Tsai <[email protected]>, Lianet Magrans <[email protected]>

Author: TaiJuWu

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

@@ -17,7 +17,7 @@
 package kafka.api
 
 import kafka.api.ConsumerRebootstrapTest._
-import kafka.server.QuorumTestHarness.getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
+import kafka.server.QuorumTestHarness.getTestQuorumAndGroupProtocolParametersAll
 import kafka.utils.{TestInfoUtils, TestUtils}
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
@@ -135,12 +135,12 @@ object ConsumerRebootstrapTest {
 
   final val RebootstrapTestName = s"${TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames}.useRebootstrapTriggerMs={2}"
   def rebootstrapTestParams: stream.Stream[Arguments] = {
-    assertEquals(1, getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly.count())
-    val args = getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
-      .findFirst().get.get
-    stream.Stream.of(
-      Arguments.of((args :+ true):_*),
-      Arguments.of((args :+ false):_*)
-    )
+    getTestQuorumAndGroupProtocolParametersAll
+      .flatMap { baseArgs =>
+        stream.Stream.of(
+          Arguments.of((baseArgs.get :+ true):_*),
+          Arguments.of((baseArgs.get :+ false):_*)
+        )
+      }
   }
 }

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

@@ -207,10 +207,10 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
     Admin.create(config)
   }
 
-  @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
-  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
   @Timeout(10)
-  def testDescribeUserScramCredentialsTimeout(quorum: String, groupProtocol: String): Unit = {
+  @ParameterizedTest
+  @ValueSource(strings = Array("kraft"))
+  def testDescribeUserScramCredentialsTimeout(quorum: String): Unit = {
     client = createInvalidAdminClient()
     try {
       // test describeUserScramCredentials(List<String> users, DescribeUserScramCredentialsOptions options)