build: detect unused parameters and local variables

codestyle/pmd-ruleset.xml

@@ -29,4 +29,6 @@ This ruleset defines the PMD rules for the Apache Druid project.
 
   <rule ref="category/java/codestyle.xml/UnnecessaryImport" />
   <rule ref="category/java/codestyle.xml/TooManyStaticImports" />
+  <rule ref="category/java/bestpractices.xml/UnusedFormalParameter" />
+  <rule ref="category/java/bestpractices.xml/UnusedLocalVariable" />
 </ruleset>

extensions-contrib/opentelemetry-emitter/src/main/java/org/apache/druid/emitter/opentelemetry/OpenTelemetryEmitter.java

@@ -82,7 +82,7 @@ private void emitQueryTimeEvent(ServiceMetricEvent event)
   {
     Context opentelemetryContext = propagator.extract(Context.current(), event, DRUID_CONTEXT_TEXT_MAP_GETTER);
 
-    try (Scope scope = opentelemetryContext.makeCurrent()) {
+    try (Scope ignoredScope = opentelemetryContext.makeCurrent()) {
       DateTime endTime = event.getCreatedTime();
       DateTime startTime = endTime.minusMillis(event.getValue().intValue());
 

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamRecordSupplier.java

@@ -275,11 +275,14 @@ private void filterBufferAndResetBackgroundFetch(Set<StreamPartition<String>> pa
   {
     this.stopBackgroundFetch();
     // filter records in buffer and only retain ones whose partition was not seeked
-    BlockingQueue<OrderedPartitionableRecord<String, Long, ByteEntity>> newQ = new LinkedBlockingQueue<>(
+    final Set<String> partitionIds = partitions.stream()
+                                               .map(StreamPartition::getPartitionId)
+                                               .collect(Collectors.toSet());
+    final BlockingQueue<OrderedPartitionableRecord<String, Long, ByteEntity>> newQ = new LinkedBlockingQueue<>(
         recordBufferSize);
 
     queue.stream()
-        .filter(x -> !streamBuilders.containsKey(x.getPartitionId()))
+        .filter(x -> !partitionIds.contains(x.getPartitionId()))
         .forEachOrdered(newQ::offer);
 
     queue = newQ;

extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamRecordSupplierTest.java

@@ -27,6 +27,7 @@
 import com.rabbitmq.stream.ConsumerBuilder;
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.EnvironmentBuilder;
+import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.MessageHandler;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.codec.WrapperMessageBuilder;
@@ -386,6 +387,51 @@ public void testSeek()
   }
 
 
+  @Test
+  public void testSeekRetainsBufferedRecordsForOtherPartitions()
+  {
+    final StreamPartition<String> partition0 = StreamPartition.of(STREAM, PARTITION_ID0);
+    final StreamPartition<String> partition1 = StreamPartition.of(STREAM, PARTITION_ID1);
+    final Set<StreamPartition<String>> partitions = ImmutableSet.of(partition0, partition1);
+    final RabbitStreamRecordSupplier recordSupplier = makeRecordSupplierWithMockedEnvironment(uri, null);
+
+    EasyMock.expect(environmentBuilder.uri("rabbitmq-stream://localhost:5552")).andReturn(environmentBuilder).once();
+    EasyMock.expect(environmentBuilder.build()).andStubReturn(environment);
+
+    final ConsumerBuilder consumerBuilder0 = createMock(ConsumerBuilder.class);
+    EasyMock.expect(environment.consumerBuilder()).andReturn(consumerBuilder0).once();
+    EasyMock.expect(consumerBuilder0.noTrackingStrategy()).andReturn(consumerBuilder0).once();
+    EasyMock.expect(consumerBuilder0.stream(PARTITION_ID0)).andReturn(consumerBuilder0).once();
+    EasyMock.expect(consumerBuilder0.messageHandler(recordSupplier)).andReturn(consumerBuilder0).once();
+
+    final ConsumerBuilder consumerBuilder1 = createMock(ConsumerBuilder.class);
+    EasyMock.expect(environment.consumerBuilder()).andReturn(consumerBuilder1).once();
+    EasyMock.expect(consumerBuilder1.noTrackingStrategy()).andReturn(consumerBuilder1).once();
+    EasyMock.expect(consumerBuilder1.stream(PARTITION_ID1)).andReturn(consumerBuilder1).once();
+    EasyMock.expect(consumerBuilder1.messageHandler(recordSupplier)).andReturn(consumerBuilder1).once();
+
+    replayAll();
+    recordSupplier.assign(partitions);
+
+    final WrapperMessageBuilder messageBuilder = new WrapperMessageBuilder();
+    messageBuilder.addData("record".getBytes(StandardCharsets.UTF_8));
+    final Message rabbitMessage = messageBuilder.build();
+    for (int i = 0; i < 50; i++) {
+      recordSupplier.handle(new MessageHandlerContext(i, 0, 0, PARTITION_ID0), rabbitMessage);
+      recordSupplier.handle(new MessageHandlerContext(i, 0, 0, PARTITION_ID1), rabbitMessage);
+    }
+
+    recordSupplier.seek(partition0, 10L);
+
+    final List<OrderedPartitionableRecord<String, Long, ByteEntity>> messages = recordSupplier.poll(0);
+    Assert.assertEquals(50, messages.size());
+    Assert.assertTrue(messages.stream().allMatch(message -> PARTITION_ID1.equals(message.getPartitionId())));
+
+    recordSupplier.close();
+    verifyAll();
+  }
+
+
 
   @Test
   public void testPollBothPartitions()

extensions-core/google-extensions/src/main/java/org/apache/druid/storage/google/GoogleTaskLogs.java

@@ -119,24 +119,24 @@ private void pushTaskFile(final File logFile, final String taskKey) throws IOExc
   public Optional<InputStream> streamTaskLog(final String taskid, final long offset) throws IOException
   {
     final String taskKey = getTaskLogKey(taskid);
-    return streamTaskFile(taskid, offset, taskKey);
+    return streamTaskFile(offset, taskKey);
   }
 
   @Override
   public Optional<InputStream> streamTaskReports(String taskid) throws IOException
   {
     final String taskKey = getTaskReportKey(taskid);
-    return streamTaskFile(taskid, 0, taskKey);
+    return streamTaskFile(0, taskKey);
   }
 
   @Override
   public Optional<InputStream> streamTaskStatus(String taskid) throws IOException
   {
     final String taskKey = getTaskStatusKey(taskid);
-    return streamTaskFile(taskid, 0, taskKey);
+    return streamTaskFile(0, taskKey);
   }
 
-  private Optional<InputStream> streamTaskFile(final String taskid, final long offset, String taskKey)
+  private Optional<InputStream> streamTaskFile(final long offset, String taskKey)
       throws IOException
   {
     try {

indexing-service/src/main/java/org/apache/druid/indexing/overlord/sampler/InputSourceSampler.java

@@ -127,7 +127,7 @@ public SamplerResponse sample(
       );
       try (final CloseableIterator<InputRowListPlusRawValues> iterator = reader.sample();
            final IncrementalIndex index = buildIncrementalIndex(nonNullSamplerConfig, nonNullDataSchema);
-           final Closer closer1 = closer) {
+           final Closer ignoredCloser = closer) {
         List<SamplerResponseRow> responseRows = new ArrayList<>(nonNullSamplerConfig.getNumRows());
         int numRowsIndexed = 0;
 

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -1924,9 +1924,6 @@ private List<ParseExceptionReport> getCurrentParseErrors()
       }
     }
 
-    SeekableStreamIndexTaskTuningConfig ss = spec.getSpec().getTuningConfig().convertToTaskTuningConfig();
-    SeekableStreamSupervisorIOConfig oo = spec.getSpec().getIOConfig();
-
     // store a limited number of parse exceptions, keeping the most recent ones
     int parseErrorLimit = spec.getSpec().getTuningConfig().convertToTaskTuningConfig().getMaxSavedParseExceptions() *
                           spec.getSpec().getIOConfig().getTaskCount();

indexing-service/src/main/java/org/apache/druid/indexing/worker/shuffle/LocalIntermediaryDataManager.java

@@ -301,7 +301,7 @@ public DataSegment addSegment(String supervisorTaskId, String subTaskId, DataSeg
     final BucketNumberedShardSpec<?> bucketNumberedShardSpec = (BucketNumberedShardSpec<?>) segment.getShardSpec();
 
     //noinspection unused
-    try (final Closer resourceCloser = closer) {
+    try (final Closer ignoredCloser = closer) {
       FileUtils.mkdirp(taskTempDir);
 
       // Temporary compressed file. Will be removed when taskTempDir is deleted.

processing/src/main/java/org/apache/druid/common/utils/SocketUtil.java

@@ -41,7 +41,7 @@ public static int findOpenPortFrom(int startPort)
     int currPort = startPort;
 
     while (currPort < 0xffff) {
-      try (ServerSocket socket = new ServerSocket(currPort)) {
+      try (ServerSocket ignoredSocket = new ServerSocket(currPort)) {
         return currPort;
       }
       catch (IOException e) {

processing/src/main/java/org/apache/druid/frame/processor/FrameProcessors.java

@@ -77,8 +77,8 @@ public void cleanup() throws IOException
       {
         if (cleanedUp.compareAndSet(false, true)) {
           //noinspection EmptyTryBlock
-          try (Closeable ignore1 = baggage;
-               Closeable ignore2 = processor::cleanup) {
+          try (Closeable ignoredBaggage = baggage;
+               Closeable ignoredCleanup = processor::cleanup) {
             // piggy-back try-with-resources semantics
           }
         }

processing/src/main/java/org/apache/druid/java/util/common/FileUtils.java

@@ -264,7 +264,7 @@ public static <T> T writeAtomically(final File file, final File tmpDir, OutputSt
     final File tmpFile = new File(tmpDir, StringUtils.format(".%s.%s", file.getName(), UUID.randomUUID()));
 
     //noinspection unused
-    try (final Closeable deleter = () -> Files.deleteIfExists(tmpFile.toPath())) {
+    try (final Closeable ignoredDeleter = () -> Files.deleteIfExists(tmpFile.toPath())) {
       final T retVal;
 
       try (

processing/src/main/java/org/apache/druid/java/util/common/guava/ConcatSequence.java

@@ -142,7 +142,7 @@ public boolean isDone()
       @Override
       public void close() throws IOException
       {
-        try (Closeable toClose = yielderYielder) {
+        try (Closeable ignoredYielder = yielderYielder) {
           yielder.close();
         }
       }

server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java

@@ -521,7 +521,7 @@ private Sink getOrCreateSink(final SegmentIdWithShardSpec identifier)
           tuningConfig.getIndexSpec(),
           Collections.emptyList()
       );
-      bytesCurrentlyInMemory.addAndGet(calculateSinkMemoryInUsed(retVal));
+      bytesCurrentlyInMemory.addAndGet(calculateSinkMemoryInUsed());
 
       // Add sink prior to announcing it, to ensure it is immediately queryable.
       addSink(identifier, retVal);
@@ -1526,7 +1526,7 @@ private ListenableFuture<?> abandonSegment(
       // i.e. those that haven't been persisted for *InMemory counters, or pushed to deep storage for the total counter.
       rowsCurrentlyInMemory.addAndGet(-sink.getNumRowsInMemory());
       bytesCurrentlyInMemory.addAndGet(-sink.getBytesInMemory());
-      bytesCurrentlyInMemory.addAndGet(-calculateSinkMemoryInUsed(sink));
+      bytesCurrentlyInMemory.addAndGet(-calculateSinkMemoryInUsed());
       for (FireHydrant hydrant : sink) {
         // Decrement memory used by all Memory Mapped Hydrant
         if (!hydrant.equals(sink.getCurrHydrant())) {
@@ -1801,7 +1801,7 @@ private int calculateMMappedHydrantMemoryInUsed(FireHydrant hydrant)
     return total;
   }
 
-  private int calculateSinkMemoryInUsed(Sink sink)
+  private int calculateSinkMemoryInUsed()
   {
     if (skipBytesInMemoryOverheadCheck) {
       return 0;

server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java

@@ -807,7 +807,7 @@ private int dropReplicas(
     // Drop as many replicas as possible from decommissioning servers
     int remainingNumToDrop = numToDrop;
     int numDropsQueued =
-        dropReplicasFromServers(remainingNumToDrop, segment, eligibleDyingServers.iterator(), tier);
+        dropReplicasFromServers(remainingNumToDrop, segment, eligibleDyingServers.iterator());
 
     // Drop replicas from active servers if required
     if (numToDrop > numDropsQueued) {
@@ -816,7 +816,7 @@ private int dropReplicas(
           (useRoundRobinAssignment || eligibleLiveServers.size() <= remainingNumToDrop)
           ? eligibleLiveServers.iterator()
           : strategy.findServersToDropSegment(segment, new ArrayList<>(eligibleLiveServers));
-      numDropsQueued += dropReplicasFromServers(remainingNumToDrop, segment, serverIterator, tier);
+      numDropsQueued += dropReplicasFromServers(remainingNumToDrop, segment, serverIterator);
     }
 
     return numDropsQueued;
@@ -829,8 +829,7 @@ private int dropReplicas(
   private int dropReplicasFromServers(
       int numToDrop,
       DataSegment segment,
-      Iterator<ServerHolder> serverIterator,
-      String tier
+      Iterator<ServerHolder> serverIterator
   )
   {
     int numDropsQueued = 0;

sql/src/main/java/org/apache/druid/sql/avatica/DruidAvaticaJsonHandler.java

@@ -69,7 +69,7 @@ public boolean handle(Request request, Response response, Callback callback) thr
     String remoteAddr = Request.getRemoteAddr(request);
     DruidMeta.setThreadLocalRemoteAddress(remoteAddr);
 
-    try (Timer.Context ctx = this.requestTimer.start()) {
+    try (Timer.Context ignoredContext = this.requestTimer.start()) {
       if (AVATICA_PATH_NO_TRAILING_SLASH.equals(StringUtils.maybeRemoveTrailingSlash(requestURI))) {
         response.getHeaders().put("Content-Type", "application/json;charset=utf-8");
 

sql/src/main/java/org/apache/druid/sql/avatica/DruidAvaticaProtobufHandler.java

@@ -73,7 +73,7 @@ public boolean handle(Request request, Response response, Callback callback) thr
 
     try {
       if (AVATICA_PATH_NO_TRAILING_SLASH.equals(StringUtils.maybeRemoveTrailingSlash(requestURI))) {
-        try (Timer.Context ctx = this.requestTimer.start()) {
+        try (Timer.Context ignoredContext = this.requestTimer.start()) {
           if (!"POST".equals(request.getMethod())) {
             response.setStatus(405);
             response.write(

sql/src/main/java/org/apache/druid/sql/calcite/rule/logical/UnnestInputCleanupRule.java

@@ -83,7 +83,7 @@ public void onMatch(RelOptRuleCall call)
 
     newProjects.set(inputIndex, null);
 
-    RexNode newUnnestExpr = unnestInput.accept(new ExpressionPullerRexShuttle(newProjects, inputIndex));
+    RexNode newUnnestExpr = unnestInput.accept(new ExpressionPullerRexShuttle(newProjects));
 
     if (newUnnestExpr instanceof RexInputRef) {
       // this won't make it simpler
@@ -142,7 +142,7 @@ private static class ExpressionPullerRexShuttle extends RexShuttle
   {
     private final List<RexNode> projects;
 
-    private ExpressionPullerRexShuttle(List<RexNode> projects, int replaceableIndex)
+    private ExpressionPullerRexShuttle(List<RexNode> projects)
     {
       this.projects = projects;
     }

sql/src/main/java/org/apache/druid/sql/calcite/schema/InformationSchema.java

@@ -377,8 +377,7 @@ public Iterable<Object[]> apply(final String tableName)
                                       return generateColumnMetadata(
                                           schemaName,
                                           tableName,
-                                           table.getRowType(typeFactory),
-                                          typeFactory
+                                          table.getRowType(typeFactory)
                                       );
                                     }
                                   }
@@ -395,8 +394,7 @@ public Iterable<Object[]> apply(final String functionName)
                                           return generateColumnMetadata(
                                               schemaName,
                                               functionName,
-                                              viewMacro.apply(Collections.emptyList()).getRowType(typeFactory),
-                                              typeFactory
+                                              viewMacro.apply(Collections.emptyList()).getRowType(typeFactory)
                                           );
                                         }
                                         catch (Exception e) {
@@ -442,8 +440,7 @@ public TableType getJdbcTableType()
     private Iterable<Object[]> generateColumnMetadata(
         final String schemaName,
         final String tableName,
-        final RelDataType tableSchema,
-        final RelDataTypeFactory typeFactory
+        final RelDataType tableSchema
     )
     {
       return FluentIterable