MINOR: add assertion about groupEpoch and targetAssignmentEpoch to testConsumerGroups

Signed-off-by: PoAn Yang <[email protected]>

Author: FrankYang0529

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

@@ -1810,7 +1810,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    * Test the consumer group APIs.
    */
   @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
-  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_KAFKA_17960"))
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
   def testConsumerGroups(quorum: String, groupProtocol: String): Unit = {
     val config = createConfig
     client = Admin.create(config)
@@ -1938,8 +1938,13 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
           // Test that we can get information about the test consumer group.
           assertTrue(describeWithFakeGroupResult.describedGroups().containsKey(testGroupId))
           var testGroupDescription = describeWithFakeGroupResult.describedGroups().get(testGroupId).get()
-          assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.groupEpoch.isEmpty)
-          assertEquals(groupType == GroupType.CLASSIC, testGroupDescription.targetAssignmentEpoch.isEmpty)
+          if (groupType == GroupType.CLASSIC) {
+            assertTrue(testGroupDescription.groupEpoch.isEmpty)
+            assertTrue(testGroupDescription.targetAssignmentEpoch.isEmpty)
+          } else {
+            assertEquals(Optional.of(3), testGroupDescription.groupEpoch)
+            assertEquals(Optional.of(3), testGroupDescription.targetAssignmentEpoch)
+          }
 
           assertEquals(testGroupId, testGroupDescription.groupId())
           assertFalse(testGroupDescription.isSimpleConsumerGroup)
@@ -2167,7 +2172,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    * Test the consumer group APIs.
    */
   @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
-  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_KAFKA_17960"))
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
   def testConsumerGroupsDeprecatedConsumerGroupState(quorum: String, groupProtocol: String): Unit = {
     val config = createConfig
     client = Admin.create(config)

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

@@ -575,8 +575,5 @@ object QuorumTestHarness {
 
   // The following parameter groups are to *temporarily* avoid bugs with the CONSUMER group protocol Consumer
   // implementation that would otherwise cause tests to fail.
-  def getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_KAFKA_17960: stream.Stream[Arguments] = getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
-  def getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_KAFKA_17961: stream.Stream[Arguments] = getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
-  def getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_KAFKA_17964: stream.Stream[Arguments] = getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
   def getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly_KAFKA_18034: stream.Stream[Arguments] = getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly
 }