Update tests

Author: alanlau28

clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java

@@ -21,7 +21,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.RebootstrapRequiredException;
-import org.apache.kafka.common.errors.UnsupportedVersionException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.message.ApiMessageType;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
@@ -241,7 +241,7 @@ public void testUnsupportedVersionDuringInternalMetadataRequest() {
         // disabling auto topic creation for versions less than 4 is not supported
         MetadataRequest.Builder builder = new MetadataRequest.Builder(topics, false, (short) 3);
         client.sendInternalMetadataRequest(builder, node.idString(), time.milliseconds());
-        assertEquals(UnsupportedVersionException.class, metadataUpdater.getAndClearFailure().getClass());
+        assertEquals(UnsupportedProtocolFieldException.class, metadataUpdater.getAndClearFailure().getClass());
     }
 
     @Test

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java

@@ -31,6 +31,7 @@
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.DisconnectException;
 import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
 import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
@@ -714,9 +715,9 @@ public void testUnsupportedVersionFromBroker(String errorMsg) {
      * REGEX_RESOLUTION_NOT_SUPPORTED_MSG only generated on the client side.
      */
     @ParameterizedTest
-    @ValueSource(strings = {CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG, REGEX_RESOLUTION_NOT_SUPPORTED_MSG})
-    public void testUnsupportedVersionFromClient(String errorMsg) {
-        mockResponseWithException(new UnsupportedVersionException(errorMsg), false);
+    @MethodSource("unsupportedVersionFromClientCases")
+    public void testUnsupportedVersionFromClient(UnsupportedVersionException thrown, String errorMsg) {
+        mockResponseWithException(thrown, false);
         ArgumentCaptor<ErrorEvent> errorEventArgumentCaptor = ArgumentCaptor.forClass(ErrorEvent.class);
         verify(backgroundEventHandler).add(errorEventArgumentCaptor.capture());
         ErrorEvent errorEvent = errorEventArgumentCaptor.getValue();
@@ -725,6 +726,13 @@ public void testUnsupportedVersionFromClient(String errorMsg) {
         clearInvocations(backgroundEventHandler);
     }
 
+    private static Stream<Arguments> unsupportedVersionFromClientCases() {
+        return Stream.of(
+            Arguments.of(new UnsupportedVersionException(CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG), CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG),
+            Arguments.of(new UnsupportedProtocolFieldException(REGEX_RESOLUTION_NOT_SUPPORTED_MSG), REGEX_RESOLUTION_NOT_SUPPORTED_MSG)
+        );
+    }
+
     private void mockResponseWithException(UnsupportedVersionException exception, boolean isFromBroker) {
         time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
         NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());

clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java

@@ -35,6 +35,7 @@
 import org.apache.kafka.common.errors.NotEnoughReplicasException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
 import org.apache.kafka.common.errors.UnknownServerException;
+import org.apache.kafka.common.errors.UnsupportedProtocolFieldException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.message.AddOffsetsToTxnRequestData;
 import org.apache.kafka.common.message.AddOffsetsToTxnResponseData;
@@ -683,8 +684,8 @@ public void testCreateTopicRequestV0FailsIfValidateOnly() {
 
     @Test
     public void testCreateTopicRequestV3FailsIfNoPartitionsOrReplicas() {
-        final UnsupportedVersionException exception = assertThrows(
-            UnsupportedVersionException.class, () -> {
+        final UnsupportedProtocolFieldException exception = assertThrows(
+            UnsupportedProtocolFieldException.class, () -> {
                 CreateTopicsRequestData data = new CreateTopicsRequestData()
                     .setTimeoutMs(123)
                     .setValidateOnly(false);
@@ -699,8 +700,7 @@ public void testCreateTopicRequestV3FailsIfNoPartitionsOrReplicas() {
 
                 new Builder(data).build((short) 3);
             });
-        assertTrue(exception.getMessage().contains("supported in CreateTopicRequest version 4+"));
-        assertTrue(exception.getMessage().contains("[foo, bar]"));
+        assertTrue(exception.getMessage().contains("does not support [foo,bar] in CREATE_TOPICS API version 3"));
     }
 
     @Test