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

Code before:
    /**
     * @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 rely on topic id only
     * otherwise the method will rely 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 (Utils.isBlank(tpId.topic()) && !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();
        return result;
    }

    @Override

Review comment (performance):
In the rare case that `Sender::topicIdsForBatches` returns 0 topic id (e.g. topic is deleted), we will pass along topicName -> 0 to handleProduceResponse(). The response will include empty topic and 0 topic id. It's important that we find a match in this case to avoid IllegalStateException. I am thinking that we should first try to do the comparison on topic name, if it's not empty. Otherwise, just do the comparison on topic id even if it's zero.

Code after:
     */
    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/clients/consumer/internals/FetchBuffer.java (Java)

Code before:
        } finally {
            lock.unlock();
        }
    }

    /**
     * Return whether we have any completed fetches pending return to the user. This method is thread-safe. Has
     * visibility for testing.
     *
     * @return {@code true} if there are completed fetches that match the {@link Predicate}, {@code false} otherwise
     */
    boolean hasCompletedFetches(Predicate<CompletedFetch> predicate) {
        try {
            lock.lock();
            return completedFetches.stream().anyMatch(predicate);
        } finally {
            lock.unlock();
        }
    }

    void add(CompletedFetch completedFetch) {
        try {
            lock.lock();
            completedFetches.add(completedFetch);
            wokenup.set(true);
            blockingCondition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    void addAll(Collection<CompletedFetch> completedFetches) {
        if (completedFetches == null || completedFetches.isEmpty())
            return;

        try {
            lock.lock();
            this.completedFetches.addAll(completedFetches);
            wokenup.set(true);
            blockingCondition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    CompletedFetch nextInLineFetch() {
        try {
            lock.lock();
            return nextInLineFetch;
        } finally {
            lock.unlock();

Review comment (suggestion):
could we reuse the existing `addAll` in this implementation of `add`? Would avoid dup logic like this

Code after:
            return completedFetches.isEmpty();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Return whether we have any completed fetches pending return to the user. This method is thread-safe. Has
     * visibility for testing.
     *
     * @return {@code true} if there are completed fetches that match the {@link Predicate}, {@code false} otherwise
     */
    boolean hasCompletedFetches(Predicate<CompletedFetch> predicate) {
        try {
            lock.lock();
            return completedFetches.stream().anyMatch(predicate);
        } finally {
            lock.unlock();
        }
    }

    void add(CompletedFetch completedFetch) {
        addAll(List.of(completedFetch));
    }

    void addAll(Collection<CompletedFetch> completedFetches) {
        if (completedFetches == null || completedFetches.isEmpty())
            return;

        try {
            lock.lock();
            this.completedFetches.addAll(completedFetches);
            wokenup.set(true);
            blockingCondition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    CompletedFetch nextInLineFetch() {
        try {
            lock.lock();
            return nextInLineFetch;
        } finally {
            lock.unlock();
        }
    }

    void setNextInLineFetch(CompletedFetch nextInLineFetch) {
        try {
            lock.lock();
Repository: apache/kafka
File: jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsStickyAssignorBenchmark.java (Java)

Code before:
            ));
        }

        updatedMemberSpec.put("newMember", new AssignmentMemberSpec(
            Optional.empty(),
            Optional.empty(),
            Map.of(),
            Map.of(),
            Map.of(),
            "process-newMember",
            Map.of(),
            Map.of(),
            Map.of()
        ));

        groupSpec = new GroupSpecImpl(
            updatedMemberSpec,
            assignmentConfigs
        );
    }

    @Benchmark
    @Threads(1)
    @OutputTimeUnit(TimeUnit.MILLISECONDS)
    public void doAssignment() {
        taskAssignor.assign(groupSpec, topologyDescriber);
    }
}


Review comment (performance):
This should either return the calculated `GroupAssignment` object or use a [JMH Blackhole](https://javadoc.io/doc/org.openjdk.jmh/jmh-core/1.23/org/openjdk/jmh/infra/Blackhole.html) to consume it so the JIT compiler thinks it's used elsewhere and doesn't get optimized away as dead code.  Probably makes more sense to just update the return type and add `return taskAssignor.assign(xxx, yyy)` 

Code after:
                Map.of()
            ));
        }

        updatedMemberSpec.put("newMember", new AssignmentMemberSpec(
            Optional.empty(),
            Optional.empty(),
            Map.of(),
            Map.of(),
            Map.of(),
            "process-newMember",
            Map.of(),
            Map.of(),
            Map.of()
        ));

        groupSpec = new GroupSpecImpl(
            updatedMemberSpec,
            assignmentConfigs
        );
    }

    @Benchmark
    @Threads(1)
    @OutputTimeUnit(TimeUnit.MILLISECONDS)
    public void doAssignment(Blackhole blackhole) {
        blackhole.consume(taskAssignor.assign(groupSpec, topologyDescriber));
    }
}
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchException.java (Java)

Code before:
 * 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.clients.consumer.internals;

import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.SerializationException;

public class ShareFetchException extends SerializationException {

    private final KafkaException origin;

    private final ShareFetch<?, ?> shareFetch;

    public ShareFetchException(KafkaException exception, ShareFetch<?, ?> shareFetch) {
        this.origin = exception;
        this.shareFetch = shareFetch;
    }

    public KafkaException origin() {
        return origin;
    }

    public ShareFetch<?, ?> shareFetch() {
        return shareFetch;
    }
}


Review comment (suggestion):
I suggest `ShareFetchException(ShareFetch<?, ?> shareFetch, KafkaException cause)`. If anything, the origin seems to be the `ShareFetch` to me.

Code after:
 * 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.clients.consumer.internals;

import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.SerializationException;

public class ShareFetchException extends SerializationException {

    private final ShareFetch<?, ?> shareFetch;

    private final KafkaException cause;

    public ShareFetchException(ShareFetch<?, ?> shareFetch, KafkaException cause) {
        this.shareFetch = shareFetch;
        this.cause = cause;
    }

    public ShareFetch<?, ?> shareFetch() {
        return shareFetch;
    }

    public KafkaException cause() {
        return cause;
    }
}
Repository: apache/kafka
File: trogdor/src/main/java/org/apache/kafka/trogdor/rest/TasksResponse.java (Java)

Code before:
 *    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.trogdor.rest;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;

import java.util.Map;
import java.util.TreeMap;

/**
 * The response to /coordinator/tasks
 */
public class TasksResponse extends Message {
    private final Map<String, TaskState> tasks;

    @JsonCreator
    public TasksResponse(@JsonProperty("tasks") TreeMap<String, TaskState> tasks) {
        this.tasks = Map.copyOf((tasks == null) ? new TreeMap<>() : tasks);
    }

    @JsonProperty
    public Map<String, TaskState> tasks() {
        return tasks;
    }
}


Review comment (suggestion):
```java
this.tasks = tasks == null ? Map.of() : Map.copyOf(tasks);
```

Code after:
 *    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.trogdor.rest;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;

import java.util.Map;
import java.util.TreeMap;

/**
 * The response to /coordinator/tasks
 */
public class TasksResponse extends Message {
    private final Map<String, TaskState> tasks;

    @JsonCreator
    public TasksResponse(@JsonProperty("tasks") TreeMap<String, TaskState> tasks) {
        this.tasks = tasks == null ? Map.of() : Map.copyOf(tasks);
    }

    @JsonProperty
    public Map<String, TaskState> tasks() {
        return tasks;
    }
}
Repository: apache/kafka
File: trogdor/src/main/java/org/apache/kafka/trogdor/task/TaskSpec.java (Java)

Code before:
    /**
     * Hydrate this task on the agent.
     *
     * @param id        The worker id.
     */
    public abstract TaskWorker newTaskWorker(String id);

    @Override
    public final boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        return toString().equals(o.toString());
    }

    @Override
    public final int hashCode() {
        return Objects.hashCode(toString());
    }

    @Override
    public String toString() {
        return JsonUtil.toJsonString(this);
    }

    protected static Map<String, String> configOrEmptyMap(Map<String, String> config) {
        return (config == null) ? Map.of() : config;
    }
}


Review comment (suggestion):
```java
return config == null ? Map.of() : config;
```

Code after:
    /**
     * Hydrate this task on the agent.
     *
     * @param id        The worker id.
     */
    public abstract TaskWorker newTaskWorker(String id);

    @Override
    public final boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        return toString().equals(o.toString());
    }

    @Override
    public final int hashCode() {
        return Objects.hashCode(toString());
    }

    @Override
    public String toString() {
        return JsonUtil.toJsonString(this);
    }

    protected static Map<String, String> configOrEmptyMap(Map<String, String> config) {
        return config == null ? Map.of() : config;
    }
}
Repository: apache/kafka
File: trogdor/src/main/java/org/apache/kafka/trogdor/workload/TopicsSpec.java (Java)

Code before:
    public static final TopicsSpec EMPTY = new TopicsSpec().immutableCopy();

    private final Map<String, PartitionsSpec> map;

    @JsonCreator
    public TopicsSpec() {
        this.map = new HashMap<>();
    }

    private TopicsSpec(Map<String, PartitionsSpec> map) {
        this.map = map;
    }

    @JsonAnyGetter
    public Map<String, PartitionsSpec> get() {
        return map;
    }

    @JsonAnySetter
    public void set(String name, PartitionsSpec value) {
        map.put(name, value);
    }

    public TopicsSpec immutableCopy() {
        HashMap<String, PartitionsSpec> mapCopy = new HashMap<>(map);
        return new TopicsSpec(Map.copyOf(mapCopy));
    }

    /**
     * Enumerate the partitions inside this TopicsSpec.
     *
     * @return      A map from topic names to PartitionsSpec objects.
     */
    public Map<String, PartitionsSpec> materialize() {
        HashMap<String, PartitionsSpec> all = new HashMap<>();
        for (Map.Entry<String, PartitionsSpec> entry : map.entrySet()) {
            String topicName = entry.getKey();
            PartitionsSpec partitions = entry.getValue();
            for (String expandedTopicName : StringExpander.expand(topicName))
                all.put(expandedTopicName, partitions);
        }
        return all;
    }
}


Review comment (suggestion):
```java
return new TopicsSpec(Map.copyOf(map));
```

Code after:
    public static final TopicsSpec EMPTY = new TopicsSpec().immutableCopy();

    private final Map<String, PartitionsSpec> map;

    @JsonCreator
    public TopicsSpec() {
        this.map = new HashMap<>();
    }

    private TopicsSpec(Map<String, PartitionsSpec> map) {
        this.map = map;
    }

    @JsonAnyGetter
    public Map<String, PartitionsSpec> get() {
        return map;
    }

    @JsonAnySetter
    public void set(String name, PartitionsSpec value) {
        map.put(name, value);
    }

    public TopicsSpec immutableCopy() {
        return new TopicsSpec(Map.copyOf(map));
    }

    /**
     * Enumerate the partitions inside this TopicsSpec.
     *
     * @return      A map from topic names to PartitionsSpec objects.
     */
    public Map<String, PartitionsSpec> materialize() {
        HashMap<String, PartitionsSpec> all = new HashMap<>();
        for (Map.Entry<String, PartitionsSpec> entry : map.entrySet()) {
            String topicName = entry.getKey();
            PartitionsSpec partitions = entry.getValue();
            for (String expandedTopicName : StringExpander.expand(topicName))
                all.put(expandedTopicName, partitions);
        }
        return all;
    }
}
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java (Java)

Code before:
        final Optional<KafkaException> error;

        if (e != null)
            error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(e, "User rebalance callback throws an error"));
        else
            error = Optional.empty();

        return new ConsumerRebalanceListenerCallbackCompletedEvent(methodName, future, error);
    }

    @Override
    public String clientId() {
        return clientId;
    }

    @Override
    public Metrics metricsRegistry() {
        return metrics;
    }

    @Override
    public KafkaConsumerMetrics kafkaConsumerMetrics() {
        return kafkaConsumerMetrics;
    }

    public AsyncConsumerMetrics asyncConsumerMetrics() {
        return asyncConsumerMetrics;
    }

    // Visible for testing
    SubscriptionState subscriptions() {
        return subscriptions;
    }

    private long defaultApiTimeoutDeadlineMs() {
        return calculateDeadlineMs(time, defaultApiTimeoutMs);
    }
}


Review comment (suggestion):
I don't think this needs to be public. The test is in the same package. For `kafkaConsumerMetrics()`, there is a test which accesses the metrics object from the external object via the delegate.

Code after:
        final Optional<KafkaException> error;

        if (e != null)
            error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(e, "User rebalance callback throws an error"));
        else
            error = Optional.empty();

        return new ConsumerRebalanceListenerCallbackCompletedEvent(methodName, future, error);
    }

    @Override
    public String clientId() {
        return clientId;
    }

    @Override
    public Metrics metricsRegistry() {
        return metrics;
    }

    @Override
    public KafkaConsumerMetrics kafkaConsumerMetrics() {
        return kafkaConsumerMetrics;
    }

    AsyncConsumerMetrics asyncConsumerMetrics() {
        return asyncConsumerMetrics;
    }

    // Visible for testing
    SubscriptionState subscriptions() {
        return subscriptions;
    }

    private long defaultApiTimeoutDeadlineMs() {
        return calculateDeadlineMs(time, defaultApiTimeoutMs);
    }
}
Repository: apache/kafka
File: raft/src/main/java/org/apache/kafka/raft/RaftManager.java (Java)

Code before:
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.raft;

import org.apache.kafka.common.Node;
import org.apache.kafka.common.network.ListenerName;
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.Optional;
import java.util.concurrent.CompletableFuture;

public interface RaftManager<T> {

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

    void register(RaftClient.Listener<T> listener);

    LeaderAndEpoch leaderAndEpoch();

    RaftClient<T> client();

    ReplicatedLog replicatedLog();

    Optional<Node> voterNode(int id, ListenerName listener);

    RecordSerde<T> recordSerde();
}


Review comment (suggestion):
`register` and `leaderAndEpoch` are also a part of `RaftClient` APIs. We could remove them from `RaftManager`, since callers can use `client` instead. 

Code after:
 * 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
    );

    /**
     * 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: core/src/main/scala/kafka/raft/KafkaRaftManager.scala (Scala)

Code before:
    val discoverBrokerVersions = true

    val networkClient = new NetworkClient(
      selector,
      new ManualMetadataUpdater(),
      clientId,
      maxInflightRequestsPerConnection,
      reconnectBackoffMs,
      reconnectBackoffMsMs,
      Selectable.USE_DEFAULT_BUFFER_SIZE,
      config.socketReceiveBufferBytes,
      config.quorumConfig.requestTimeoutMs,
      config.connectionSetupTimeoutMs,
      config.connectionSetupTimeoutMaxMs,
      time,
      discoverBrokerVersions,
      apiVersions,
      logContext,
      MetadataRecoveryStrategy.NONE
    )

    (controllerListenerName, networkClient)
  }

  override def voterNode(id: Int, listener: ListenerName): Optional[Node] = {
    client.voterNode(id, listener)
  }

  override def recordSerde: RecordSerde[T] = serde
}


Review comment (suggestion):
If the delegation is only a single line and the `client` is an interface, perhaps we don't need to wrap it with `RaftManager`

Code after:

    val networkClient = new NetworkClient(
      selector,
      new ManualMetadataUpdater(),
      clientId,
      maxInflightRequestsPerConnection,
      reconnectBackoffMs,
      reconnectBackoffMsMs,
      Selectable.USE_DEFAULT_BUFFER_SIZE,
      config.socketReceiveBufferBytes,
      config.quorumConfig.requestTimeoutMs,
      config.connectionSetupTimeoutMs,
      config.connectionSetupTimeoutMaxMs,
      time,
      discoverBrokerVersions,
      apiVersions,
      logContext,
      MetadataRecoveryStrategy.NONE
    )

    (controllerListenerName, networkClient)
  }

  override def recordSerde: RecordSerde[T] = serde
}