ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/kafka
File: raft/src/main/java/org/apache/kafka/raft/RaftManager.java (Java)

Code before:
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.raft;

import org.apache.kafka.common.protocol.ApiMessage;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.server.common.serialization.RecordSerde;

import java.util.concurrent.CompletableFuture;

public interface RaftManager<T> {

    CompletableFuture<ApiMessage> handleRequest(
        RequestContext context,
        RequestHeader header,
        ApiMessage request,
        long createdTimeMs
    );

    RaftClient<T> client();

    ReplicatedLog replicatedLog();

    RecordSerde<T> recordSerde();
}


Review comment (suggestion):
Please add comments to remind developers that those methods return the same closable objects. Therefore, callers should NOT close them

Code after:
import org.apache.kafka.server.common.serialization.RecordSerde;

import java.util.concurrent.CompletableFuture;

public interface RaftManager<T> {

    CompletableFuture<ApiMessage> handleRequest(
        RequestContext context,
        RequestHeader header,
        ApiMessage request,
        long createdTimeMs
    );

    /**
     * Returns a Raft client.
     * <p>
     * Always returns the same instance. Callers must NOT close it.
     */
    RaftClient<T> client();

    /**
     * Returns a replicated log.
     * <p>
     * Always returns the same instance. Callers must NOT close it.
     */
    ReplicatedLog replicatedLog();

    /**
     * Returns the record Serde.
     */
    RecordSerde<T> recordSerde();
}
Repository: apache/kafka
File: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/ConnectRestServer.java (Java)

Code before:
    private final RestClient restClient;
    private Herder herder;

    public ConnectRestServer(Integer rebalanceTimeoutMs, RestClient restClient, Map<?, ?> props) {
        super(RestServerConfig.forPublic(rebalanceTimeoutMs, props));
        this.restClient = restClient;
    }

    public void initializeResources(Herder herder) {
        this.herder = herder;
        super.initializeResources();
    }

    @Override
    protected Collection<Class<?>> regularResources() {
        return List.of(
                RootResource.class,
                ConnectorsResource.class,
                InternalConnectResource.class,
                ConnectorPluginsResource.class
        );
    }

    @Override
    protected Collection<Class<?>> adminResources() {
        return List.of(
                LoggingResource.class
        );
    }

    @Override
    protected void configureRegularResources(ResourceConfig resourceConfig) {
        registerRestExtensions(herder, resourceConfig);
        resourceConfig.register(new Binder());
    }

    private class Binder extends AbstractBinder {
        @Override
        protected void configure() {
            bind(herder).to(Herder.class);
            bind(restClient).to(RestClient.class);
            bind(config).to(RestServerConfig.class);
        }
    }

    @Override
    protected void configureAdminResources(ResourceConfig resourceConfig) {
        resourceConfig.register(new Binder());
    }

}

Review comment (suggestion):
```java
return List.of(LoggingResource.class);
```

Code after:
    private final RestClient restClient;
    private Herder herder;

    public ConnectRestServer(Integer rebalanceTimeoutMs, RestClient restClient, Map<?, ?> props) {
        super(RestServerConfig.forPublic(rebalanceTimeoutMs, props));
        this.restClient = restClient;
    }

    public void initializeResources(Herder herder) {
        this.herder = herder;
        super.initializeResources();
    }

    @Override
    protected Collection<Class<?>> regularResources() {
        return List.of(
                RootResource.class,
                ConnectorsResource.class,
                InternalConnectResource.class,
                ConnectorPluginsResource.class
        );
    }

    @Override
    protected Collection<Class<?>> adminResources() {
        return List.of(LoggingResource.class);
    }

    @Override
    protected void configureRegularResources(ResourceConfig resourceConfig) {
        registerRestExtensions(herder, resourceConfig);
        resourceConfig.register(new Binder());
    }

    private class Binder extends AbstractBinder {
        @Override
        protected void configure() {
            bind(herder).to(Herder.class);
            bind(restClient).to(RestClient.class);
            bind(config).to(RestServerConfig.class);
        }
    }

    @Override
    protected void configureAdminResources(ResourceConfig resourceConfig) {
        resourceConfig.register(new Binder());
    }

}
Repository: apache/kafka
File: transaction-coordinator/src/test/java/org/apache/kafka/coordinator/transaction/TransactionLogTest.java (Java)

Code before:

        // Read the buffer with readTxnRecordValue.
        buffer.rewind();
        var metadata = TransactionLog.readTxnRecordValue("transaction-id", buffer);

        assertFalse(metadata == null, "Expected transaction metadata but got none");
        assertEquals(1000L, metadata.producerId());
        assertEquals(100, metadata.producerEpoch());
        assertEquals(1000, metadata.txnTimeoutMs());
        assertEquals(TransactionState.COMPLETE_COMMIT, metadata.state());
        assertEquals(Set.of(new TopicPartition("topic", 1)), metadata.topicPartitions());
        assertEquals(2000L, metadata.txnLastUpdateTimestamp());
        assertEquals(3000L, metadata.txnStartTimestamp());
    }

    @Test
    void testReadTxnRecordKeyThrowsOnUnknownVersion() {
        var unknownRecord = MessageUtil.toVersionPrefixedBytes(Short.MAX_VALUE, new TransactionLogKey());
        var exception = assertThrows(IllegalStateException.class,
            () -> TransactionLog.readTxnRecordKey(wrap(unknownRecord)));
        assertTrue(exception.getMessage().contains("Unknown version " + Short.MAX_VALUE));
    }
   
    @Test
    void shouldReturnNullForTombstoneRecord() {
        assertEquals(null, TransactionLog.readTxnRecordValue("transaction-id", null));
    }
}

Review comment (suggestion):
```java
assertNull(TransactionLog.readTxnRecordValue("transaction-id", null));
```

Code after:
        assertEquals(List.of(100, 101), value.transactionPartitions().get(0).unknownTaggedFields().stream().map(RawTaggedField::tag).toList());

        // Read the buffer with readTxnRecordValue.
        buffer.rewind();
        var metadata = TransactionLog.readTxnRecordValue("transaction-id", buffer);

        assertNotNull(metadata, "Expected transaction metadata but got none");
        assertEquals(1000L, metadata.producerId());
        assertEquals(100, metadata.producerEpoch());
        assertEquals(1000, metadata.txnTimeoutMs());
        assertEquals(TransactionState.COMPLETE_COMMIT, metadata.state());
        assertEquals(Set.of(new TopicPartition("topic", 1)), metadata.topicPartitions());
        assertEquals(2000L, metadata.txnLastUpdateTimestamp());
        assertEquals(3000L, metadata.txnStartTimestamp());
    }

    @Test
    void testReadTxnRecordKeyThrowsOnUnknownVersion() {
        var unknownRecord = MessageUtil.toVersionPrefixedBytes(Short.MAX_VALUE, new TransactionLogKey());
        var exception = assertThrows(IllegalStateException.class,
            () -> TransactionLog.readTxnRecordKey(wrap(unknownRecord)));
        assertTrue(exception.getMessage().contains("Unknown version " + Short.MAX_VALUE));
    }
   
    @Test
    void shouldReturnNullForTombstoneRecord() {
        assertNull(TransactionLog.readTxnRecordValue("transaction-id", null));
    }
}
Repository: apache/kafka
File: core/src/main/scala/kafka/server/KafkaRequestHandler.scala (Scala)

Code before:
    val currentSize = threadPoolSize.get
    info(s"Resizing request handler thread pool size from $currentSize to $newSize")
    if (newSize > currentSize) {
      for (i <- currentSize until newSize) {
        createHandler(i)
      }
    } else if (newSize < currentSize) {
      for (i <- 1 to (currentSize - newSize)) {
        runnables.remove(currentSize - i).stop()
        aggregateThreads.getAndDecrement()
      }
    }
    threadPoolSize.set(newSize)
  }

  def shutdown(): Unit = synchronized {
    info("shutting down")
    for (handler <- runnables)
      handler.initiateShutdown()
    for (handler <- runnables)
      handler.awaitShutdown()
    // Unregister this pool's threads from shared aggregate counter
    aggregateThreads.addAndGet(-threadPoolSize.get)
    info("shut down completely")
  }
}


Review comment (refactor):
It is an inconsistent design that the aggregate threads are incremented through `createHandler` but they are decremented through `resizeThreadPool`. You should be able to remove this inconsistency by fixing the constructor and calling `resizeThreadPool` in the constructor.

You can also enforce this by making `createHandler` private and removing the `synchronized`. You can also add a private method (internalResizeThreadPool) that doesn't `synchronized` which is call by the constructor and `resizeThreadPool` which does `synchronized`.

Code after:
  }
  /* Per-pool idle meter (broker-only or controller-only) */
  private val perPoolIdleMeter = metricsGroup.newMeter(perPoolIdleMeterName, "percent", TimeUnit.NANOSECONDS)
  /* Aggregate meter to track the average free capacity of the request handlers */
  private val aggregateIdleMeter = metricsGroup.newMeter(requestHandlerAvgIdleMetricName, "percent", TimeUnit.NANOSECONDS)

  this.logIdent = s"[data-plane Kafka Request Handler on ${nodeName.capitalize} $brokerId] "
  val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)
  for (i <- 0 until numThreads) {
    createHandler(i)
  }

  private def createHandler(id: Int): Unit = {
    runnables += new KafkaRequestHandler(
      id,
      brokerId,
      aggregateIdleMeter,
      aggregateThreads,
      perPoolIdleMeter,
      threadPoolSize,
      requestChannel,
      apis,
      time,
      nodeName
    )
    aggregateThreads.getAndIncrement()
    KafkaThread.daemon("data-plane-kafka-request-handler-" + id, runnables(id)).start()
  }

  private def deleteHandler(id: Int): Unit = {
    runnables.remove(id).stop()
    aggregateThreads.getAndDecrement()
  }

  def resizeThreadPool(newSize: Int): Unit = synchronized {
    val currentSize = threadPoolSize.get
    info(s"Resizing request handler thread pool size from $currentSize to $newSize")
    if (newSize > currentSize) {
      for (i <- currentSize until newSize) {
        createHandler(i)
      }
    } else if (newSize < currentSize) {
      for (i <- 1 to (currentSize - newSize)) {
        deleteHandler(currentSize - i)
      }
    }
    threadPoolSize.set(newSize)
  }

  def shutdown(): Unit = synchronized {
    info("shutting down")
Repository: apache/kafka
File: core/src/main/scala/kafka/server/KafkaRequestHandler.scala (Scala)

Code before:
    completeShutdown()
  }

  private def completeShutdown(): Unit = {
    requestLocal.close()
    threadRequestChannel.remove()
    shutdownComplete.countDown()
  }

  def stop(): Unit = {
    stopped = true
  }

  def initiateShutdown(): Unit = requestChannel.sendShutdownRequest()

  def awaitShutdown(): Unit = shutdownComplete.await()

}

/**
 * Factory for creating KafkaRequestHandlerPool instances with shared aggregate metrics.
 * All pools created by the same factory share the same aggregateThreads counter.
 */
class KafkaRequestHandlerPoolFactory {
  val aggregateThreads = new AtomicInteger(0)
  val RequestHandlerAvgIdleMetricName = "RequestHandlerAvgIdlePercent"
  
  def createPool(
    brokerId: Int,
    requestChannel: RequestChannel,
    apis: ApiRequestHandler,
    time: Time,
    numThreads: Int,
    nodeName: String
  ): KafkaRequestHandlerPool = {
    new KafkaRequestHandlerPool(aggregateThreads, RequestHandlerAvgIdleMetricName, brokerId, requestChannel, apis, time, numThreads, nodeName)
  }
}

class KafkaRequestHandlerPool(
  val aggregateThreads: AtomicInteger,
  val requestHandlerAvgIdleMetricName: String,
  val brokerId: Int,
  val requestChannel: RequestChannel,
  val apis: ApiRequestHandler,
  time: Time,
  numThreads: Int,
  nodeName: String
) extends Logging {
  private val metricsGroup = new KafkaMetricsGroup(this.getClass)


Review comment (suggestion):
Make this fields private with:
```scala
  private[this] val aggregateThreads = new AtomicInteger(0)
  private[this] val RequestHandlerAvgIdleMetricName = "RequestHandlerAvgIdlePercent"
```

Code after:
    completeShutdown()
  }

  private def completeShutdown(): Unit = {
    requestLocal.close()
    threadRequestChannel.remove()
    shutdownComplete.countDown()
  }

  def stop(): Unit = {
    stopped = true
  }

  def initiateShutdown(): Unit = requestChannel.sendShutdownRequest()

  def awaitShutdown(): Unit = shutdownComplete.await()

}

/**
 * Factory for creating KafkaRequestHandlerPool instances with shared aggregate metrics.
 * All pools created by the same factory share the same aggregateThreads counter.
 */
class KafkaRequestHandlerPoolFactory {
  private[this] val aggregateThreads = new AtomicInteger(0)
  private[this] val RequestHandlerAvgIdleMetricName = "RequestHandlerAvgIdlePercent"
  
  def createPool(
    brokerId: Int,
    requestChannel: RequestChannel,
    apis: ApiRequestHandler,
    time: Time,
    numThreads: Int,
    nodeName: String
  ): KafkaRequestHandlerPool = {
    new KafkaRequestHandlerPool(aggregateThreads, RequestHandlerAvgIdleMetricName, brokerId, requestChannel, apis, time, numThreads, nodeName)
  }

  // Only used for test purpose
  def aggregateThreadCount: Int = aggregateThreads.get()
}

class KafkaRequestHandlerPool(
  val aggregateThreads: AtomicInteger,
  val requestHandlerAvgIdleMetricName: String,
  val brokerId: Int,
  val requestChannel: RequestChannel,
  val apis: ApiRequestHandler,
  time: Time,
  numThreads: Int,
  nodeName: String
Repository: apache/kafka
File: raft/src/main/java/org/apache/kafka/raft/EpochState.java (Java)

Code before:

    /**
     * Get the current election state, which is guaranteed to be immutable.
     */
    ElectionState election();

    /**
     * Get the current (immutable) epoch.
     */
    int epoch();

    /**
     * Returns the known endpoints for the leader.
     *
     * If the leader is not known then {@code Endpoints.empty()} is returned.
     */
    Endpoints leaderEndpoints();

    /**
     * User-friendly description of the state
     */
    String name();

    @Override
    default void close() {
        // Since all subclasses implement the Closeable interface while none throw any IOException,
        // this default implementation is provided to eliminate the need for exception handling in the close operation.
    }
}


Review comment (suggestion):
It feels odd to add a default implementation just for the sake of  documentation. How about using the following style instead?
```java
    /**
     * Since all subclasses implement the Closeable interface while none throw any IOException,
     * this default implementation is provided to eliminate the need for exception handling in the close operation.
     */
    @Override
    void close();
```

Code after:

    /**
     * Get the current election state, which is guaranteed to be immutable.
     */
    ElectionState election();

    /**
     * Get the current (immutable) epoch.
     */
    int epoch();

    /**
     * Returns the known endpoints for the leader.
     *
     * If the leader is not known then {@code Endpoints.empty()} is returned.
     */
    Endpoints leaderEndpoints();

    /**
     * User-friendly description of the state
     */
    String name();

    /**
     * Since all subclasses implement the Closeable interface while none throw any IOException,
     * this implementation is provided to eliminate the need for exception handling in the close operation.
     */
    @Override
    void close();
}
Repository: apache/kafka
File: metadata/src/main/java/org/apache/kafka/metadata/publisher/FeaturesPublisher.java (Java)

Code before:

    @Override
    public String name() {
        return "FeaturesPublisher";
    }

    @Override
    public void onMetadataUpdate(
        MetadataDelta delta,
        MetadataImage newImage,
        LoaderManifest manifest
    ) {
        try {
            if (delta.featuresDelta() != null) {
                FinalizedFeatures newFinalizedFeatures = new FinalizedFeatures(newImage.features().metadataVersionOrThrow(),
                        newImage.features().finalizedVersions(),
                        newImage.provenance().lastContainedOffset()
                );
                if (!newFinalizedFeatures.equals(finalizedFeatures)) {
                    log.info("Loaded new metadata {}.", newFinalizedFeatures);
                    finalizedFeatures = newFinalizedFeatures;
                }
            }
        } catch (Throwable t) {
            String deltaName = "MetadataDelta up to " + newImage.highestOffsetAndEpoch().offset();
            faultHandler.handleFault("Uncaught exception while publishing features changes from " + deltaName, t);
        }
    }
}


Review comment (suggestion):
we could concat the strings directly, right? for example:
```java
            faultHandler.handleFault("Uncaught exception while publishing SCRAM changes from MetadataDelta up to "
                + newImage.highestOffsetAndEpoch().offset(), t);
```

Code after:

    @Override
    public String name() {
        return "FeaturesPublisher";
    }

    @Override
    public void onMetadataUpdate(
        MetadataDelta delta,
        MetadataImage newImage,
        LoaderManifest manifest
    ) {
        try {
            if (delta.featuresDelta() != null) {
                FinalizedFeatures newFinalizedFeatures = new FinalizedFeatures(newImage.features().metadataVersionOrThrow(),
                    newImage.features().finalizedVersions(),
                    newImage.provenance().lastContainedOffset()
                );
                if (!newFinalizedFeatures.equals(finalizedFeatures)) {
                    log.info("Loaded new metadata {}.", newFinalizedFeatures);
                    finalizedFeatures = newFinalizedFeatures;
                }
            }
        } catch (Throwable t) {
            faultHandler.handleFault("Uncaught exception while publishing SCRAM changes from MetadataDelta up to "
                + newImage.highestOffsetAndEpoch().offset(), t);
        }
    }
}
Repository: apache/kafka
File: metadata/src/test/java/org/apache/kafka/metadata/MetadataVersionConfigValidatorTest.java (Java)

Code before:
        verifyNoMoreInteractions(faultHandler);

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_7_IV2, false, faultHandler);
        verifyNoMoreInteractions(faultHandler);

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_6_IV2, true, faultHandler);
        verify(faultHandler, times(1)).handleFault(
            eq("Broker configuration does not support the cluster MetadataVersion"),
            any(IllegalArgumentException.class));

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_7_IV0, true, faultHandler);
        verify(faultHandler, times(1)).handleFault(
            eq("Broker configuration does not support the cluster MetadataVersion"),
            any(IllegalArgumentException.class));

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_7_IV2, true, faultHandler);
        verifyNoMoreInteractions(faultHandler);
    }

    private void validate(MetadataVersion metadataVersion, boolean jbodConfig, FaultHandler faultHandler)
        throws Exception {
        Supplier<Boolean> multiLogDirSupplier = mock(Supplier.class);
        when(multiLogDirSupplier.get()).thenReturn(jbodConfig);

        testWith(metadataVersion, multiLogDirSupplier, faultHandler);
    }
}


Review comment (suggestion):
```java
        Supplier<Boolean> multiLogDirSupplier = () -> jbodConfig;
```

Code after:

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_7_IV2, false, faultHandler);
        verifyNoMoreInteractions(faultHandler);

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_6_IV2, true, faultHandler);
        verify(faultHandler, times(1)).handleFault(
            eq("Broker configuration does not support the cluster MetadataVersion"),
            any(IllegalArgumentException.class));

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_7_IV0, true, faultHandler);
        verify(faultHandler, times(1)).handleFault(
            eq("Broker configuration does not support the cluster MetadataVersion"),
            any(IllegalArgumentException.class));

        faultHandler = mock(FaultHandler.class);
        validate(MetadataVersion.IBP_3_7_IV2, true, faultHandler);
        verifyNoMoreInteractions(faultHandler);
    }

    private void validate(MetadataVersion metadataVersion, boolean jbodConfig, FaultHandler faultHandler)
        throws Exception {
        Supplier<Boolean> multiLogDirSupplier = () -> jbodConfig;

        executeMetadataUpdate(metadataVersion, multiLogDirSupplier, faultHandler);
    }
}