ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/AllocateProducerIdsRequestTest.scala (Scala)

Code before:

    val allocateResponse = sendAndReceiveAllocateProducerIds(sourceBroker, controllerServer)
    assertEquals(Errors.NOT_CONTROLLER, Errors.forCode(allocateResponse.data.errorCode))
  }

  private def sendAndReceiveAllocateProducerIds(
    sourceBroker: BrokerServer,
    controllerServer: ControllerServer
  ): AllocateProducerIdsResponse = {
    val allocateRequest = new AllocateProducerIdsRequest.Builder(
      new AllocateProducerIdsRequestData()
        .setBrokerId(sourceBroker.config.brokerId)
        .setBrokerEpoch(sourceBroker.lifecycleManager.brokerEpoch)
    ).build()

    connectAndReceive(
      controllerServer.socketServer,
      allocateRequest
    )
  }

  private def connectAndReceive(
    controllerSocketServer: SocketServer,
    request: AllocateProducerIdsRequest
  ): AllocateProducerIdsResponse = {

    val listenerName = cluster.controllerListenerName
    val port = controllerSocketServer.boundPort(listenerName)

    IntegrationTestUtils.connectAndReceive[AllocateProducerIdsResponse](
      request,
      port
    )
  }

}


Review comment (suggestion):
```scala
    IntegrationTestUtils.connectAndReceive[AllocateProducerIdsResponse](
      request,
      controllerSocketServer.boundPort(cluster.controllerListenerName)
    )
```

Code after:

    val allocateResponse = sendAndReceiveAllocateProducerIds(sourceBroker, controllerServer)
    assertEquals(Errors.NOT_CONTROLLER, Errors.forCode(allocateResponse.data.errorCode))
  }

  private def sendAndReceiveAllocateProducerIds(
    sourceBroker: BrokerServer,
    controllerServer: ControllerServer
  ): AllocateProducerIdsResponse = {
    val allocateRequest = new AllocateProducerIdsRequest.Builder(
      new AllocateProducerIdsRequestData()
        .setBrokerId(sourceBroker.config.brokerId)
        .setBrokerEpoch(sourceBroker.lifecycleManager.brokerEpoch)
    ).build()

    connectAndReceive(
      controllerServer.socketServer,
      allocateRequest
    )
  }

  private def connectAndReceive(
    controllerSocketServer: SocketServer,
    request: AllocateProducerIdsRequest
  ): AllocateProducerIdsResponse = {
    IntegrationTestUtils.connectAndReceive[AllocateProducerIdsResponse](
      request,
      controllerSocketServer.boundPort(cluster.controllerListenerName())
    )
  }
}
Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/test/MockDeserializer.java (Java)

Code before:
    public boolean isKey;
    public Map<String, ?> configs;

    public static void resetStaticVariables() {
        initCount = new AtomicInteger(0);
        closeCount = new AtomicInteger(0);
        clusterMeta = new AtomicReference<>();
        clusterIdBeforeDeserialize = new AtomicReference<>(noClusterId);
    }

    public MockDeserializer() {
        initCount.incrementAndGet();
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.configs = configs;
        this.isKey = isKey;
    }

    @Override
    public String deserialize(String topic, byte[] data) {
        // This will ensure that we get the cluster metadata when deserialize is called for the first time
        // as subsequent compareAndSet operations will fail.
        clusterIdBeforeDeserialize.compareAndSet(noClusterId, clusterMeta.get());
        if (data == null)
            return null;
        else
            return new String(data, StandardCharsets.UTF_8);
    }

    @Override
    public void close() {
        closeCount.incrementAndGet();
    }

    @Override
    public void onUpdate(ClusterResource clusterResource) {
        clusterMeta.set(clusterResource);
    }
}

Review comment (suggestion):
```java
        if (data == null) return null;
        return data.getBytes(StandardCharsets.UTF_8);
```

Code after:
    public boolean isKey;
    public Map<String, ?> configs;

    public static void resetStaticVariables() {
        initCount = new AtomicInteger(0);
        closeCount = new AtomicInteger(0);
        clusterMeta = new AtomicReference<>();
        clusterIdBeforeDeserialize = new AtomicReference<>(noClusterId);
    }

    public MockDeserializer() {
        initCount.incrementAndGet();
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.configs = configs;
        this.isKey = isKey;
    }

    @Override
    public String deserialize(String topic, byte[] data) {
        // This will ensure that we get the cluster metadata when deserialize is called for the first time
        // as subsequent compareAndSet operations will fail.
        clusterIdBeforeDeserialize.compareAndSet(noClusterId, clusterMeta.get());
        if (data == null) return null;
        return new String(data, StandardCharsets.UTF_8);
    }

    @Override
    public void close() {
        closeCount.incrementAndGet();
    }

    @Override
    public void onUpdate(ClusterResource clusterResource) {
        clusterMeta.set(clusterResource);
    }
}
Repository: apache/kafka
File: core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala (Scala)

Code before:
  private def buildFetchParams(replicaId: Int,
                               maxWaitMs: Int): FetchParams = {
    new FetchParams(
      replicaId,
      1,
      maxWaitMs,
      1,
      maxBytes,
      FetchIsolation.LOG_END,
      Optional.empty()
    )
  }

  private def buildReadResult(error: Errors,
                              highWatermark: Int = 0,
                              leaderLogStartOffset: Int = 0): LogReadResult = {
    new LogReadResult(
      new FetchDataInfo(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, MemoryRecords.EMPTY),
      Optional.empty(),
      highWatermark,
      leaderLogStartOffset,
      -1L,
      -1L,
      -1L,
      OptionalLong.empty(),
      if (error != Errors.NONE) Optional.of(error.exception).asInstanceOf[Optional[Throwable]] else Optional.empty[Throwable]())
  }

}


Review comment (suggestion):
Will this be better?
```suggestion
      if (error != Errors.NONE) Optional.of[Throwable](error.exception) else Optional.empty[Throwable]())
```

Code after:
  private def buildFetchParams(replicaId: Int,
                               maxWaitMs: Int): FetchParams = {
    new FetchParams(
      replicaId,
      1,
      maxWaitMs,
      1,
      maxBytes,
      FetchIsolation.LOG_END,
      Optional.empty()
    )
  }

  private def buildReadResult(error: Errors,
                              highWatermark: Int = 0,
                              leaderLogStartOffset: Int = 0): LogReadResult = {
    new LogReadResult(
      new FetchDataInfo(LogOffsetMetadata.UNKNOWN_OFFSET_METADATA, MemoryRecords.EMPTY),
      Optional.empty(),
      highWatermark,
      leaderLogStartOffset,
      -1L,
      -1L,
      -1L,
      OptionalLong.empty(),
      if (error != Errors.NONE) Optional.of[Throwable](error.exception) else Optional.empty[Throwable]())
  }

}
Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java (Java)

Code before:
                    return true;
                } finally {
                    Utils.closeQuietly(raf, "index file " + file.getName());
                }
            }
        });
    }

    /**
     * Rename the file that backs this offset index
     *
     * @throws IOException if rename fails
     */
    public void renameTo(File f) throws IOException {
        try {
            Utils.atomicMoveWithFallback(file.toPath(), f.toPath(), false);
        } finally {
            this.file = f;
        }
    }

    /**
     * Flush the data in the index to disk
     */
    public void flush() {
        LockUtils.inLock(lock, () -> {
            mmap.force();
        });
    }

    /**
     * Delete this index file.
     *
     * @throws IOException if deletion fails due to an I/O error
     * @return `true` if the file was deleted by this method; `false` if the file could not be deleted because it did
     *         not exist
     */
    public boolean deleteIfExists() throws IOException {
        closeHandler();
        return Files.deleteIfExists(file.toPath());
    }

    /**
     * Trim this segment to fit just the valid entries, deleting all trailing unwritten bytes from
     * the file.
     */
    public void trimToValidSize() throws IOException {
        LockUtils.inLockThrows(lock, () -> {
            if (mmap != null) {
                resize(entrySize() * entries);
            }

Review comment (suggestion):
This could just be `inLock( () -> {`, right? Ditto in a few other places.

Code after:
                                    mmap.position(), mmap.limit());
                            return true;
                        } finally {
                            Utils.closeQuietly(raf, "index file " + file.getName());
                        }
                    }
                }));
    }

    /**
     * Rename the file that backs this offset index
     *
     * @throws IOException if rename fails
     */
    public void renameTo(File f) throws IOException {
        try {
            Utils.atomicMoveWithFallback(file.toPath(), f.toPath(), false);
        } finally {
            this.file = f;
        }
    }

    /**
     * Flush the data in the index to disk
     */
    public void flush() {
        inLock(() -> {
            mmap.force();
        });
    }

    /**
     * Delete this index file.
     *
     * @throws IOException if deletion fails due to an I/O error
     * @return `true` if the file was deleted by this method; `false` if the file could not be deleted because it did
     *         not exist
     */
    public boolean deleteIfExists() throws IOException {
        closeHandler();
        return Files.deleteIfExists(file.toPath());
    }

    /**
     * Trim this segment to fit just the valid entries, deleting all trailing unwritten bytes from
     * the file.
     */
    public void trimToValidSize() throws IOException {
        inLockThrows(() -> {
            if (mmap != null) {
                resize(entrySize() * entries);
Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/OffsetIndex.java (Java)

Code before:

    private int relativeOffset(ByteBuffer buffer, int n) {
        return buffer.getInt(n * ENTRY_SIZE);
    }

    private int physical(ByteBuffer buffer, int n) {
        return buffer.getInt(n * ENTRY_SIZE + 4);
    }

    /**
     * Truncates index to a known number of entries.
     */
    private void truncateToEntries(int entries) {
        inLock(() -> {
            super.truncateToEntries0(entries);
            this.lastOffset = lastEntry().offset;
            log.debug("Truncated index {} to {} entries; position is now {} and last offset is now {}",
                file().getAbsolutePath(), entries, mmap().position(), lastOffset);
        });
    }

    /**
     * The last entry in the index
     */
    private OffsetPosition lastEntry() {
        return inLock(() -> {
            int entries = entries();
            if (entries == 0)
                return new OffsetPosition(baseOffset(), 0);
            else
                return parseEntry(mmap(), entries - 1);
        });
    }
}


Review comment (suggestion):
lastEntry() is a reader. So it should only take remap read lock.

Also, there is an existing issue. It seems there is no memory barrier for the instance level field lastOffset. So a reader may not see the latest value. We need to make it volatile.

Code after:

    private int relativeOffset(ByteBuffer buffer, int n) {
        return buffer.getInt(n * ENTRY_SIZE);
    }

    private int physical(ByteBuffer buffer, int n) {
        return buffer.getInt(n * ENTRY_SIZE + 4);
    }

    /**
     * Truncates index to a known number of entries.
     */
    private void truncateToEntries(int entries) {
        inLock(() -> {
            super.truncateToEntries0(entries);
            this.lastOffset = lastEntry().offset;
            log.debug("Truncated index {} to {} entries; position is now {} and last offset is now {}",
                file().getAbsolutePath(), entries, mmap().position(), lastOffset);
        });
    }

    /**
     * The last entry in the index
     */
    private OffsetPosition lastEntry() {
        return inRemapReadLock(() -> {
            int entries = entries();
            if (entries == 0)
                return new OffsetPosition(baseOffset(), 0);
            else
                return parseEntry(mmap(), entries - 1);
        });
    }
}
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java (Java)

Code before:
    public String topic() {
        return topicPartition.topic();
    }

    /**
     * @return the partition id.
     */
    public int partition() {
        return topicPartition.partition();
    }

    /**
     * @return Topic partition representing this instance.
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    /**
     * Checking if TopicIdPartition meant to be the same reference to same this object but doesn't have all the data.
     * If topic name is empty and topic id is persisted then the method will relay on topic id only
     * otherwise the method will relay on topic name.
     * @return true if topic has same topicId and partition index as topic names some time might be empty.
    */
    public boolean same(TopicIdPartition tpId) {
        if (tpId.topic().isEmpty() && !tpId.topicId.equals(Uuid.ZERO_UUID)) {
            return topicId.equals(tpId.topicId) &&
                    topicPartition.partition() == tpId.partition();
        } else {
            return topicPartition.equals(tpId.topicPartition());
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TopicIdPartition that = (TopicIdPartition) o;
        return topicId.equals(that.topicId) &&
               topicPartition.equals(that.topicPartition);
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = prime + topicId.hashCode();
        result = prime * result + topicPartition.hashCode();

Review comment (suggestion):
`TopicIdPartition#topic` is nullable, so the null check is necessary.

Code after:
    public String topic() {
        return topicPartition.topic();
    }

    /**
     * @return the partition id.
     */
    public int partition() {
        return topicPartition.partition();
    }

    /**
     * @return Topic partition representing this instance.
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TopicIdPartition that = (TopicIdPartition) o;
        return topicId.equals(that.topicId) &&
               topicPartition.equals(that.topicPartition);
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = prime + topicId.hashCode();
        result = prime * result + topicPartition.hashCode();
        return result;
    }

    @Override
    public String toString() {
        return topicId + ":" + topic() + "-" + partition();
    }
}
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java (Java)

Code before:

    /**
     * @return the topic name or null if it is unknown.
     */
    public String topic() {
        return topicPartition.topic();
    }

    /**
     * @return the partition id.
     */
    public int partition() {
        return topicPartition.partition();
    }

    /**
     * @return Topic partition representing this instance.
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    /**
     * Checking if TopicIdPartition meant to be the same reference to same this object but doesn't have all the data.
     * If topic name is empty and topic id is persisted then the method will relay on topic id only
     * otherwise the method will relay on topic name.
     * @return true if topic has same topicId and partition index as topic names some time might be empty.
    */
    public boolean same(TopicIdPartition tpId) {
        boolean emptyTopicName = tpId.topic() == null || tpId.topic().isEmpty();
        if (emptyTopicName && !tpId.topicId.equals(Uuid.ZERO_UUID)) {
            return topicId.equals(tpId.topicId) &&
                    topicPartition.partition() == tpId.partition();
        } else {
            return topicPartition.equals(tpId.topicPartition());
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TopicIdPartition that = (TopicIdPartition) o;
        return topicId.equals(that.topicId) &&
               topicPartition.equals(that.topicPartition);
    }


Review comment (suggestion):
Nit: I believe this is meant to be _rely_:

```suggestion
     * If topic name is empty and topic id is persisted then the method will rely on topic id only
     * otherwise the method will rely on topic name.
```

Code after:

    /**
     * @return the topic name or null if it is unknown.
     */
    public String topic() {
        return topicPartition.topic();
    }

    /**
     * @return the partition id.
     */
    public int partition() {
        return topicPartition.partition();
    }

    /**
     * @return Topic partition representing this instance.
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TopicIdPartition that = (TopicIdPartition) o;
        return topicId.equals(that.topicId) &&
               topicPartition.equals(that.topicPartition);
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = prime + topicId.hashCode();
        result = prime * result + topicPartition.hashCode();
        return result;
    }

    @Override
    public String toString() {
        return topicId + ":" + topic() + "-" + partition();
    }
}
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java (Java)

Code before:
    public String topic() {
        return topicPartition.topic();
    }

    /**
     * @return the partition id.
     */
    public int partition() {
        return topicPartition.partition();
    }

    /**
     * @return Topic partition representing this instance.
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    /**
     * Checking if TopicIdPartition meant to be the same reference to same this object but doesn't have all the data.
     * If topic name is empty and topic id is persisted then the method will relay on topic id only
     * otherwise the method will relay on topic name.
     * @return true if topic has same topicId and partition index as topic names some time might be empty.
    */
    public boolean same(TopicIdPartition tpId) {
        boolean emptyTopicName = tpId.topic() == null || tpId.topic().isEmpty();
        if (emptyTopicName && !tpId.topicId.equals(Uuid.ZERO_UUID)) {
            return topicId.equals(tpId.topicId) &&
                    topicPartition.partition() == tpId.partition();
        } else {
            return topicPartition.equals(tpId.topicPartition());
        }
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TopicIdPartition that = (TopicIdPartition) o;
        return topicId.equals(that.topicId) &&
               topicPartition.equals(that.topicPartition);
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = prime + topicId.hashCode();

Review comment (suggestion):
Minor suggestion—`Utils.isBlank()` is a little more succinct and catches cases where the string is whitespace only.

```suggestion
        boolean emptyTopicName = Utils.isBlank(tpId.topic());
```

Code after:
    public String topic() {
        return topicPartition.topic();
    }

    /**
     * @return the partition id.
     */
    public int partition() {
        return topicPartition.partition();
    }

    /**
     * @return Topic partition representing this instance.
     */
    public TopicPartition topicPartition() {
        return topicPartition;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        TopicIdPartition that = (TopicIdPartition) o;
        return topicId.equals(that.topicId) &&
               topicPartition.equals(that.topicPartition);
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = prime + topicId.hashCode();
        result = prime * result + topicPartition.hashCode();
        return result;
    }

    @Override
    public String toString() {
        return topicId + ":" + topic() + "-" + partition();
    }
}