impl

Author: lucasbru

checkstyle/checkstyle.xml

@@ -43,6 +43,7 @@
       <property name="tokens" value="IDENT, NUM_DOUBLE, LAMBDA, TEXT_BLOCK_LITERAL_BEGIN, UNARY_MINUS, UNARY_PLUS, INC, DEC, POST_INC, POST_DEC" />
     </module>
     <module name="SimplifyBooleanReturn"/>
+    <module name="UnusedLocalVariable"/>
 
     <!-- style -->
     <module name="DefaultComesLast"/>

clients/src/main/java/org/apache/kafka/common/requests/DescribeUserScramCredentialsRequest.java

@@ -68,7 +68,8 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
                 .setThrottleTimeMs(throttleTimeMs)
                 .setErrorCode(apiError.error().code())
                 .setErrorMessage(apiError.message());
-        for (DescribeUserScramCredentialsRequestData.UserName user : data.users()) {
+
+        for (@SuppressWarnings("UnusedLocalVariable") DescribeUserScramCredentialsRequestData.UserName ignored : data.users()) {
             response.results().add(new DescribeUserScramCredentialsResponseData.DescribeUserScramCredentialsResult()
                     .setErrorCode(apiError.error().code())
                     .setErrorMessage(apiError.message()));

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

@@ -445,7 +445,7 @@ public void testHeartbeatRequestFailureNotifiedToGroupManagerAfterErrorPropagate
         time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
         NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
         assertEquals(1, result.unsentRequests.size());
-        ClientResponse response = createHeartbeatResponse(result.unsentRequests.get(0), Errors.GROUP_AUTHORIZATION_FAILED);
+        createHeartbeatResponse(result.unsentRequests.get(0), Errors.GROUP_AUTHORIZATION_FAILED);
         result.unsentRequests.get(0).handler().onFailure(time.milliseconds(), new AuthenticationException("Fatal error in HB"));
 
         // The error should be propagated before notifying the group manager. This ensures that the app thread is aware

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

@@ -2881,7 +2881,7 @@ private void verifySessionPartitions() {
                                 field.setAccessible(true);
                                 LinkedHashMap<?, ?> sessionPartitions =
                                         (LinkedHashMap<?, ?>) field.get(handler);
-                                for (Map.Entry<?, ?> entry : sessionPartitions.entrySet()) {
+                                for (@SuppressWarnings("UnusedLocalVariable") Map.Entry<?, ?> ignored : sessionPartitions.entrySet()) {
                                     // If `sessionPartitions` are modified on another thread, Thread.yield will increase the
                                     // possibility of ConcurrentModificationException if appropriate synchronization is not used.
                                     Thread.yield();

clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java

@@ -397,7 +397,7 @@ public void testStressfulSituation() throws Exception {
             List<ProducerBatch> batches = accum.drain(metadataCache, nodes, 5 * 1024, 0).get(node1.id());
             if (batches != null) {
                 for (ProducerBatch batch : batches) {
-                    for (Record record : batch.records().records())
+                    for (@SuppressWarnings("UnusedLocalVariable") Record ignored : batch.records().records())
                         read++;
                     accum.deallocate(batch);
                 }

clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java

@@ -1674,7 +1674,7 @@ public void testForceCloseWithProducerIdReset() throws Exception {
         Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, 10,
             senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);
 
-        Future<RecordMetadata> failedResponse = appendToAccumulator(tp0);
+        appendToAccumulator(tp0);
         Future<RecordMetadata> successfulResponse = appendToAccumulator(tp1);
         sender.runOnce();  // connect and send.
 

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

@@ -527,11 +527,6 @@ public void produceRequestGetErrorResponseTest() {
     public void fetchResponseVersionTest() {
         Uuid id = Uuid.randomUuid();
         MemoryRecords records = MemoryRecords.readableRecords(ByteBuffer.allocate(10));
-        FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
-                .setPartitionIndex(0)
-                .setHighWatermark(1000000)
-                .setLogStartOffset(-1)
-                .setRecords(records);
         LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> idResponseData = new LinkedHashMap<>();
         idResponseData.put(new TopicIdPartition(id, new TopicPartition("test", 0)),
                 new FetchResponseData.PartitionData()

connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskTest.java

@@ -20,7 +20,6 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Header;
@@ -283,8 +282,6 @@ public void testCommitRecordWithNullMetadata() {
 
         @SuppressWarnings("unchecked")
         KafkaConsumer<byte[], byte[]> consumer = mock(KafkaConsumer.class);
-        @SuppressWarnings("unchecked")
-        KafkaProducer<byte[], byte[]> producer = mock(KafkaProducer.class);
         MirrorSourceMetrics metrics = mock(MirrorSourceMetrics.class);
 
         String sourceClusterName = "cluster1";

connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java

@@ -60,7 +60,7 @@ public void start(Map<String, String> props) {
         @Override
         public void put(Collection<SinkRecord> records) {
             super.put(records);
-            for (SinkRecord ignore : records) {
+            for (@SuppressWarnings("UnusedLocalVariable") SinkRecord ignore : records) {
                 count++;
             }
         }

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

@@ -751,9 +751,6 @@ public void testSendRecordsFailedTransformationErrorToleranceAll() {
 
         expectConvertHeadersAndKeyValue(emptyHeaders(), TOPIC);
 
-        TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, null, Collections.emptyList(), Collections.emptyList());
-        TopicDescription topicDesc = new TopicDescription(TOPIC, false, Collections.singletonList(topicPartitionInfo));
-
         workerTask.toSend = Arrays.asList(record1);
 
         // The transformation errored out so the error should be ignored & the record skipped with error tolerance all

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

@@ -621,6 +621,7 @@ protected void assertInitializedMetric(WorkerConnector workerConnector, String e
         String type = metrics.currentMetricValueAsString(metricGroup, "connector-type");
         String clazz = metrics.currentMetricValueAsString(metricGroup, "connector-class");
         String version = metrics.currentMetricValueAsString(metricGroup, "connector-version");
+        assertEquals("unassigned", status);
         assertEquals(expectedType, type);
         assertNotNull(clazz);
         assertEquals(VERSION, version);

core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -960,6 +960,7 @@ public void copyLogSegmentsToRemote(UnifiedLog log) throws InterruptedException
                                     // back to the caller. It's important to note that the task being executed is already
                                     // cancelled before the executing thread is interrupted. The caller is responsible
                                     // for handling the exception gracefully by checking if the task is already cancelled.
+                                    @SuppressWarnings("UnusedLocalVariable")
                                     boolean ignored = copyQuotaManagerLockCondition.await(quotaTimeout().toMillis(), TimeUnit.MILLISECONDS);
                                     throttleTimeMs = rlmCopyQuotaManager.getThrottleTimeMs();
                                 }

generator/src/main/java/org/apache/kafka/message/checker/EvolutionVerifier.java

@@ -74,7 +74,7 @@ static void verifyVersionsMatchTopLevelMessage(
             verifyVersionsMatchTopLevelMessage(what, topLevelMessage, field);
         }
         for (StructSpec struct : topLevelMessage.commonStructs()) {
-            for (FieldSpec field : topLevelMessage.fields()) {
+            for (FieldSpec field : struct.fields()) {
                 verifyVersionsMatchTopLevelMessage(what, topLevelMessage, field);
             }
         }

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/OffsetMetadataManagerTest.java

@@ -2966,7 +2966,7 @@ public void testOffsetCommitsSensor() {
         assertEquals(1, group.generationId());
         group.transitionTo(ClassicGroupState.STABLE);
 
-        CoordinatorResult<OffsetCommitResponseData, CoordinatorRecord> result = context.commitOffset(
+        context.commitOffset(
             new OffsetCommitRequestData()
                 .setGroupId("foo")
                 .setMemberId("member")

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TopicMetadataTest.java

@@ -21,10 +21,6 @@
 
 import org.junit.jupiter.api.Test;
 
-import java.util.Map;
-import java.util.Set;
-
-import static org.apache.kafka.coordinator.group.GroupCoordinatorRecordHelpersTest.mkMapOfPartitionRacks;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -49,7 +45,6 @@ public void testTopicIdAndNameCannotBeNull() {
     @Test
     public void testEquals() {
         Uuid topicId = Uuid.randomUuid();
-        Map<Integer, Set<String>> partitionRacks = mkMapOfPartitionRacks(15);
         TopicMetadata topicMetadata = new TopicMetadata(topicId, "foo", 15);
 
         assertEquals(new TopicMetadata(topicId, "foo", 15), topicMetadata);

metadata/src/test/java/org/apache/kafka/controller/QuorumControllerMetricsIntegrationTest.java

@@ -198,7 +198,7 @@ public void testEventQueueOperationsStartedMetric() throws Throwable {
                                                      build()
         ) {
             QuorumController active = controlEnv.activeController();
-            Map<Integer, Long> brokerEpochs = registerBrokersAndUnfence(active, 3);
+            registerBrokersAndUnfence(active, 3);
 
             // Test that a new operation increments operationsStarted. We retry this if needed
             // to handle the case where another operation is performed in between loading

metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java

@@ -1535,7 +1535,6 @@ public void testDeleteTopicsWithMutationQuotaExceeded() {
             anonymousContextFor(ApiKeys.CREATE_TOPICS);
         ControllerResult<CreateTopicsResponseData> createResult =
             replicationControl.createTopics(createTopicsRequestContext, request, Collections.singleton("foo"));
-        CreateTopicsResponseData expectedResponse = new CreateTopicsResponseData();
         CreatableTopicResult createdTopic = createResult.response().topics().find("foo");
         assertEquals(NONE.code(), createdTopic.errorCode());
         ctx.replay(createResult.records());

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java

@@ -189,24 +189,6 @@ public void testLeaderDoesNotBootstrapRecordsWithKraftVersion0() throws Exceptio
             .withUnknownLeader(0)
             .build();
 
-        List<List<ControlRecord>> expectedBootstrapRecords = Arrays.asList(
-            Arrays.asList(
-                new ControlRecord(
-                    ControlRecordType.SNAPSHOT_HEADER,
-                    new SnapshotHeaderRecord()
-                        .setVersion((short) 0)
-                        .setLastContainedLogTimestamp(0)
-                )
-            ),
-            Arrays.asList(
-                new ControlRecord(
-                    ControlRecordType.SNAPSHOT_FOOTER,
-                    new SnapshotFooterRecord()
-                        .setVersion((short) 0)
-                )
-            )
-        );
-
         // check leader does not write bootstrap records to log
         context.unattachedToLeader();
 

server-common/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java

@@ -653,9 +653,8 @@ public void testFindCoordinatorSuccess() {
 
         CompletableFuture<ReadShareGroupStateResponse> resultFuture = handler.result();
 
-        ReadShareGroupStateResponse result = null;
         try {
-            result = resultFuture.get();
+            resultFuture.get();
         } catch (Exception e) {
             fail("Failed to get result from future", e);
         }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogLoader.java

@@ -169,6 +169,7 @@ public LoadedLogOffsets load() throws IOException {
                     long offset = LogFileUtils.offsetFromFile(file);
                     if (offset >= minSwapFileOffset && offset < maxSwapFileOffset) {
                         logger.info("Deleting segment files {} that is compacted but has not been deleted yet.", file.getName());
+                        @SuppressWarnings("UnusedLocalVariable")
                         boolean ignore = file.delete();
                     }
                 }
@@ -186,6 +187,7 @@ public LoadedLogOffsets load() throws IOException {
             }
             if (file.getName().endsWith(LogFileUtils.SWAP_FILE_SUFFIX)) {
                 logger.info("Recovering file {} by renaming from {} files.", file.getName(), LogFileUtils.SWAP_FILE_SUFFIX);
+                @SuppressWarnings("UnusedLocalVariable")
                 boolean ignore = file.renameTo(new File(Utils.replaceSuffix(file.getPath(), LogFileUtils.SWAP_FILE_SUFFIX, "")));
             }
         }

storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java

@@ -144,7 +144,6 @@ public void teardown() throws IOException {
     })
     public void testAppendForLogSegmentOffsetOverflowException(long baseOffset, long largestOffset) throws IOException {
         try (LogSegment seg = createSegment(baseOffset, 10, Time.SYSTEM)) {
-            long currentTime = Time.SYSTEM.milliseconds();
             MemoryRecords memoryRecords = v1Records(0, "hello");
             assertThrows(LogSegmentOffsetOverflowException.class, () -> seg.append(largestOffset, memoryRecords));
         }

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KTableEfficientRangeQueryTest.java

@@ -26,7 +26,6 @@
 import org.apache.kafka.streams.TestInputTopic;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.TopologyTestDriver;
-import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
 import org.apache.kafka.streams.state.KeyValueIterator;
@@ -117,8 +116,7 @@ public void testStoreConfig(final StoreType storeType, final boolean enableLoggi
         final Materialized<String, String, KeyValueStore<Bytes, byte[]>> stateStoreConfig = getStoreConfig(storeType, TABLE_NAME, enableLogging, enableCaching);
         //Create topology: table from input topic
         final StreamsBuilder builder = new StreamsBuilder();
-        final KTable<String, String> table =
-                builder.table("input", stateStoreConfig);
+        builder.table("input", stateStoreConfig);
         final Topology topology = builder.build();
 
         try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {

streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java

@@ -212,7 +212,7 @@ public void testRegexRecordsAreProcessedAfterNewTopicCreatedWithMultipleSubtopol
 
             final StreamsBuilder builder = new StreamsBuilder();
             final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));
-            final KStream<String, String> otherStream = builder.stream(Pattern.compile("not-a-match"));
+            builder.stream(Pattern.compile("not-a-match"));
 
             pattern1Stream
                 .selectKey((k, v) -> k)

streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java

@@ -885,7 +885,8 @@ public Set<StreamTask> drainRestoredActiveTasks(final Duration timeout) {
                 restoredActiveTasksLock.lock();
                 try {
                     while (restoredActiveTasks.isEmpty() && now <= deadline) {
-                        final boolean elapsed = restoredActiveTasksCondition.await(deadline - now, TimeUnit.MILLISECONDS);
+                        @SuppressWarnings("UnusedLocalVariable")
+                        final boolean ignored = restoredActiveTasksCondition.await(deadline - now, TimeUnit.MILLISECONDS);
                         now = time.milliseconds();
                     }
                     result.addAll(restoredActiveTasks);

streams/src/test/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilderTest.java

@@ -472,7 +472,6 @@ public void shouldMarkStreamStreamJoinAsSelfJoinSingleStream() {
         final GraphNode join = getNodeByType(builder.root, StreamStreamJoinNode.class, new HashSet<>());
         assertNotNull(join);
         assertTrue(((StreamStreamJoinNode) join).getSelfJoin());
-        final GraphNode parent = join.parentNodes().stream().findFirst().get();
         final AtomicInteger count = new AtomicInteger();
         countJoinWindowNodes(count, builder.root, new HashSet<>());
         assertEquals(count.get(), 1);

streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java

@@ -34,7 +34,6 @@
 import org.apache.kafka.streams.kstream.Consumed;
 import org.apache.kafka.streams.kstream.Grouped;
 import org.apache.kafka.streams.kstream.KGroupedStream;
-import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.Named;
@@ -816,7 +815,7 @@ public void shouldWorkWithCogroupedTimeWindows() {
 
         final KGroupedStream<String, String> stream1 = builder.stream("one", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
         final KGroupedStream<String, String> stream2 = builder.stream("two", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
-        final KStream<Windowed<String>, Object> cogrouped = stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
+        stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
             .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(15)))
             .aggregate(() -> "", Named.as("test"), Materialized.as("store"))
             .suppress(Suppressed.untilWindowCloses(unbounded()))
@@ -829,7 +828,7 @@ public void shouldWorkWithCogroupedSlidingWindows() {
 
         final KGroupedStream<String, String> stream1 = builder.stream("one", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
         final KGroupedStream<String, String> stream2 = builder.stream("two", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
-        final KStream<Windowed<String>, Object> cogrouped = stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
+        stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
             .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(15)))
             .aggregate(() -> "", Named.as("test"), Materialized.as("store"))
             .suppress(Suppressed.untilWindowCloses(unbounded()))
@@ -842,7 +841,7 @@ public void shouldWorkWithCogroupedSessionWindows() {
 
         final KGroupedStream<String, String> stream1 = builder.stream("one", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
         final KGroupedStream<String, String> stream2 = builder.stream("two", Consumed.with(Serdes.String(), Serdes.String())).groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
-        final KStream<Windowed<String>, Object> cogrouped = stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
+        stream1.cogroup((key, value, aggregate) -> aggregate + value).cogroup(stream2, (key, value, aggregate) -> aggregate + value)
             .windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(15), Duration.ofMinutes(5)))
             .aggregate(() -> "", (k, v1, v2) -> "", Named.as("test"), Materialized.as("store"))
             .suppress(Suppressed.untilWindowCloses(unbounded()))

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

@@ -44,7 +44,6 @@ public void shouldNotAllowNullHeaders() {
 
     @Test
     public void shouldEstimateNullTopicAndEmptyHeadersAsZeroLength() {
-        final Headers headers = new RecordHeaders();
         final ProcessorRecordContext context = new ProcessorRecordContext(
             42L,
             73L,