MINOR: suppress build warning (#20424)

Suppress build warning.

Reviewers: TengYao Chi <frankvicky@apache.org>, Ken Huang
<s7133700@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/ClientsTestUtils.java

@@ -311,7 +311,7 @@ public class ClientsTestUtils {
         Optional<Map<TopicPartition, OffsetAndMetadata>> offsetsOpt
     ) throws InterruptedException {
 
-        var commitCallback = new RetryCommitCallback(consumer, offsetsOpt);
+        var commitCallback = new RetryCommitCallback<>(consumer, offsetsOpt);
         sendAsyncCommit(consumer, commitCallback, offsetsOpt);
 
         TestUtils.waitForCondition(() -> {

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/MetricsDuringTopicCreationDeletionTest.java

@@ -137,6 +137,7 @@ public class MetricsDuringTopicCreationDeletionTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     private Gauge<Integer> getGauge(String metricName) {
         return KafkaYammerMetrics.defaultRegistry().allMetrics().entrySet().stream()
             .filter(entry -> entry.getKey().getName().endsWith(metricName))

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/ClientTelemetryTest.java

@@ -123,6 +123,7 @@ public class ClientTelemetryTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT})
     public void testIntervalMsParser(ClusterInstance clusterInstance) {
         List<String> alterOpts = asList("--bootstrap-server", clusterInstance.bootstrapServers(),
@@ -152,6 +153,7 @@ public class ClientTelemetryTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     private static String[] toArray(List<String>... lists) {
         return Stream.of(lists).flatMap(List::stream).toArray(String[]::new);
     }

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTestUtils.java

@@ -168,6 +168,7 @@ public class WorkerTestUtils {
                 "Wrong rebalance delay in " + assignment);
     }
 
+    @SuppressWarnings("unchecked")
     public static <T, R extends ConnectRecord<R>> TransformationChain<T, R> getTransformationChain(
             RetryWithToleranceOperator<T> toleranceOperator,
             List<Object> results) {

core/src/test/java/kafka/server/share/DelayedShareFetchTest.java

@@ -1278,6 +1278,7 @@ public class DelayedShareFetchTest {
         assertEquals(1, delayedShareFetch.expiredRequestMeter().count());
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testRemoteStorageFetchTryCompleteReturnsFalse() {
         ReplicaManager replicaManager = mock(ReplicaManager.class);
@@ -1357,6 +1358,7 @@ public class DelayedShareFetchTest {
         delayedShareFetch.lock().unlock();
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testRemoteStorageFetchPartitionLeaderChanged() {
         ReplicaManager replicaManager = mock(ReplicaManager.class);
@@ -1421,6 +1423,7 @@ public class DelayedShareFetchTest {
         delayedShareFetch.lock().unlock();
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testRemoteStorageFetchTryCompleteThrowsException() {
         ReplicaManager replicaManager = mock(ReplicaManager.class);
@@ -1507,6 +1510,7 @@ public class DelayedShareFetchTest {
         delayedShareFetch.lock().unlock();
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testRemoteStorageFetchTryCompletionDueToBrokerBecomingOffline() {
         ReplicaManager replicaManager = mock(ReplicaManager.class);
@@ -2197,6 +2201,7 @@ public class DelayedShareFetchTest {
         return mock(BiConsumer.class);
     }
 
+    @SuppressWarnings("unchecked")
     static class DelayedShareFetchBuilder {
         private ShareFetch shareFetch = mock(ShareFetch.class);
         private ReplicaManager replicaManager = mock(ReplicaManager.class);

core/src/test/java/kafka/server/share/SharePartitionManagerTest.java

@@ -2470,7 +2470,6 @@ public class SharePartitionManagerTest {
     }
 
     @Test
-    @SuppressWarnings("unchecked")
     public void testShareFetchProcessingExceptions() throws Exception {
         String groupId = "grp";
         TopicIdPartition tp0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
@@ -3232,6 +3231,7 @@ public class SharePartitionManagerTest {
         return CollectionConverters.asScala(logReadResults).toSeq();
     }
 
+    @SuppressWarnings("unchecked")
     static void mockReplicaManagerDelayedShareFetch(ReplicaManager replicaManager,
                                                     DelayedOperationPurgatory<DelayedShareFetch> delayedShareFetchPurgatory) {
         doAnswer(invocationOnMock -> {

core/src/test/java/kafka/server/share/SharePartitionTest.java

@@ -110,6 +110,7 @@ import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+@SuppressWarnings("resource")
 public class SharePartitionTest {
 
     private static final String ACQUISITION_LOCK_NEVER_GOT_RELEASED = "Acquisition lock never got released.";
@@ -7991,6 +7992,7 @@ public class SharePartitionTest {
         assertNotNull(sharePartition.cachedState().get(50L).offsetState().get(69L).acquisitionLockTimeoutTask());
     }
 
+    @SuppressWarnings({"rawtypes", "unchecked"})
     @Test
     public void testContainsAbortMarker() {
         SharePartition sharePartition = SharePartitionBuilder.builder()

metadata/src/test/java/org/apache/kafka/controller/metrics/ControllerMetadataMetricsTest.java

@@ -160,6 +160,7 @@ public class ControllerMetadataMetricsTest {
         );
     }
 
+    @SuppressWarnings("unchecked") // suppress warning about Gauge typecast
     @Test
     public void testBrokerRegistrationStateMetrics() {
         MetricsRegistry registry = new MetricsRegistry();

metadata/src/test/java/org/apache/kafka/controller/metrics/QuorumControllerMetricsTest.java

@@ -164,6 +164,7 @@ public class QuorumControllerMetricsTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testTimeSinceLastHeartbeatReceivedMs() {
         MetricsRegistry registry = new MetricsRegistry();

server-common/src/test/java/org/apache/kafka/server/util/CommandLineUtilsTest.java

@@ -55,8 +55,8 @@ public class CommandLineUtilsTest {
         List<String> argArray = List.of("my.empty.property=", "my.empty.property1");
         Properties props = CommandLineUtils.parseKeyValueArgs(argArray);
 
-        assertEquals(props.getProperty("my.empty.property"), "", "Value of a key with missing value should be an empty string");
+        assertEquals("", props.getProperty("my.empty.property"), "Value of a key with missing value should be an empty string");
-        assertEquals(props.getProperty("my.empty.property1"), "", "Value of a key with missing value with no delimiter should be an empty string");
+        assertEquals("", props.getProperty("my.empty.property1"), "Value of a key with missing value with no delimiter should be an empty string");
     }
 
     @Test
@@ -64,7 +64,7 @@ public class CommandLineUtilsTest {
         List<String> argArray = List.of("my.property=value");
         Properties props = CommandLineUtils.parseKeyValueArgs(argArray);
 
-        assertEquals(props.getProperty("my.property"), "value", "Value of a single property should be 'value'");
+        assertEquals("value", props.getProperty("my.property"), "Value of a single property should be 'value'");
     }
 
     @Test
@@ -72,8 +72,8 @@ public class CommandLineUtilsTest {
         List<String> argArray = List.of("first.property=first", "second.property=second");
         Properties props = CommandLineUtils.parseKeyValueArgs(argArray);
 
-        assertEquals(props.getProperty("first.property"), "first", "Value of first property should be 'first'");
+        assertEquals("first", props.getProperty("first.property"), "Value of first property should be 'first'");
-        assertEquals(props.getProperty("second.property"), "second", "Value of second property should be 'second'");
+        assertEquals("second", props.getProperty("second.property"), "Value of second property should be 'second'");
     }
 
     @Test
@@ -81,9 +81,9 @@ public class CommandLineUtilsTest {
         List<String> argArray = List.of("first.property==first", "second.property=second=", "third.property=thi=rd");
         Properties props = CommandLineUtils.parseKeyValueArgs(argArray);
 
-        assertEquals(props.getProperty("first.property"), "=first", "Value of first property should be '=first'");
+        assertEquals("=first", props.getProperty("first.property"), "Value of first property should be '=first'");
-        assertEquals(props.getProperty("second.property"), "second=", "Value of second property should be 'second='");
+        assertEquals("second=", props.getProperty("second.property"), "Value of second property should be 'second='");
-        assertEquals(props.getProperty("third.property"), "thi=rd", "Value of second property should be 'thi=rd'");
+        assertEquals("thi=rd", props.getProperty("third.property"), "Value of second property should be 'thi=rd'");
     }
 
     Properties props = new Properties();
@@ -272,6 +272,7 @@ public class CommandLineUtilsTest {
                     Optional.of("127.0.0.2:9094"), Optional.of("127.0.0.3:9095"))).getMessage());
     }
 
+    @SuppressWarnings("unchecked")
     private OptionSpec<String> createMockOptionSpec(String name) {
         OptionSpec<String> spec = mock(OptionSpec.class);
         when(spec.toString()).thenReturn("[" + name.replaceAll("--", "") + "]");

server/src/test/java/org/apache/kafka/server/quota/ClientQuotasRequestTest.java

@@ -479,6 +479,7 @@ public class ClientQuotasRequestTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     private void testMatchEntities(ClientQuotaFilter filter, int expectedMatchSize, Predicate<ClientQuotaEntity> partition)
         throws InterruptedException {
         TestUtils.retryOnExceptionWithTimeout(5000L, () -> {

storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java

@@ -60,6 +60,7 @@ import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+@SuppressWarnings("resource")
 @ClusterTestDefaults(brokers = 3)
 public class TopicBasedRemoteLogMetadataManagerTest {
     private static final int SEG_SIZE = 1048576;
@@ -106,6 +107,7 @@ public class TopicBasedRemoteLogMetadataManagerTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     @ClusterTest
     public void testDoesTopicExistWithAdminClientExecutionError() throws ExecutionException, InterruptedException {
         // Create a mock Admin client that throws an ExecutionException (not UnknownTopicOrPartitionException)

storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerTest.java

@@ -381,6 +381,7 @@ public class RemoteLogManagerTest {
         assertFalse(remoteStorageManagerConfig.containsKey("remote.storage.manager.y"));
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     void testRemoteLogMetadataManagerWithEndpointConfig() {
         ArgumentCaptor<Map<String, Object>> capture = ArgumentCaptor.forClass(Map.class);
@@ -391,6 +392,7 @@ public class RemoteLogManagerTest {
         assertEquals(brokerId, capture.getValue().get(ServerConfigs.BROKER_ID_CONFIG));
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     void testRemoteLogMetadataManagerWithEndpointConfigOverridden() throws IOException {
         Properties props = new Properties();
@@ -430,6 +432,7 @@ public class RemoteLogManagerTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     void testConfigure() {
         ArgumentCaptor<Map<String, Object>> capture = ArgumentCaptor.forClass(Map.class);
@@ -2071,6 +2074,7 @@ public class RemoteLogManagerTest {
         assertEquals(expected, actual);
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testRemoteSizeData() {
         Supplier<RemoteLogManager.RetentionSizeData>[] invalidRetentionSizeData =
@@ -2087,6 +2091,7 @@ public class RemoteLogManagerTest {
         }
     }
 
+    @SuppressWarnings("unchecked")
     @Test
     public void testRemoteSizeTime() {
         Supplier<RemoteLogManager.RetentionTimeData>[] invalidRetentionTimeData =

streams/src/test/java/org/apache/kafka/streams/processor/internals/RecordCollectorTest.java

@@ -1961,6 +1961,7 @@ public class RecordCollectorTest {
 
                     }
 
+                    @SuppressWarnings("deprecation")
                     @Override
                     public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context, final ProducerRecord<byte[], byte[]> record, final Exception exception) {
                         sourceRawData[0] = context.sourceRawKey();
@@ -1978,15 +1979,14 @@ public class RecordCollectorTest {
         assertNull(sourceRawData[1]);
     }
 
-
     @Test
     public void shouldHaveRawDataDuringExceptionInSerialization() {
         final byte[][] sourceRawData = new byte[][]{new byte[]{}, new byte[]{}};
         try (final ErrorStringSerializer errorSerializer = new ErrorStringSerializer()) {
             final RecordCollector collector = newRecordCollector(
                     new ProductionExceptionHandler() {
+                        @SuppressWarnings({"deprecation", "rawtypes"})
                         @Override
-                        @SuppressWarnings({"rawtypes", "unused"})
                         public ProductionExceptionHandlerResponse handleSerializationException(final ErrorHandlerContext context, final ProducerRecord record, final Exception exception, final SerializationExceptionOrigin origin) {
                             sourceRawData[0] = context.sourceRawKey();
                             sourceRawData[1] = context.sourceRawValue();