ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListenerTest.java (Java)

Code before:

    @Test
    void testOnAllTasksLost() {
        final StreamsRebalanceData streamsRebalanceData = mock(StreamsRebalanceData.class);
        when(streamsRebalanceData.subtopologies()).thenReturn(Map.of());
        createRebalanceListenerWithRebalanceData(streamsRebalanceData);
        
        assertDoesNotThrow(() -> defaultStreamsRebalanceListener.onAllTasksLost());
        
        final InOrder inOrder = inOrder(taskManager, streamsRebalanceData);
        inOrder.verify(taskManager).handleLostAll();
        inOrder.verify(streamsRebalanceData).setReconciledAssignment(StreamsRebalanceData.Assignment.EMPTY);
    }

    @Test
    void testOnAllTasksLostWithException() {
        final Exception exception = new RuntimeException("sample exception");
        doThrow(exception).when(taskManager).handleLostAll();

        final StreamsRebalanceData streamsRebalanceData = mock(StreamsRebalanceData.class);
        when(streamsRebalanceData.subtopologies()).thenReturn(Map.of());
        createRebalanceListenerWithRebalanceData(streamsRebalanceData);

        final Exception actualException = assertThrows(RuntimeException.class, () -> defaultStreamsRebalanceListener.onAllTasksLost());

        assertEquals(exception,  actualException);
        verify(taskManager).handleLostAll();
        verify(streamsRebalanceData, never()).setReconciledAssignment(any());
    }
}


Review comment (suggestion):
[nitpick] Extra space before 'actualException' parameter. Should be 'assertEquals(exception, actualException);' for consistent formatting.

Code after:

    @Test
    void testOnAllTasksLost() {
        final StreamsRebalanceData streamsRebalanceData = mock(StreamsRebalanceData.class);
        when(streamsRebalanceData.subtopologies()).thenReturn(Map.of());
        createRebalanceListenerWithRebalanceData(streamsRebalanceData);
        
        assertDoesNotThrow(() -> defaultStreamsRebalanceListener.onAllTasksLost());
        
        final InOrder inOrder = inOrder(taskManager, streamsRebalanceData);
        inOrder.verify(taskManager).handleLostAll();
        inOrder.verify(streamsRebalanceData).setReconciledAssignment(StreamsRebalanceData.Assignment.EMPTY);
    }

    @Test
    void testOnAllTasksLostWithException() {
        final Exception exception = new RuntimeException("sample exception");
        doThrow(exception).when(taskManager).handleLostAll();

        final StreamsRebalanceData streamsRebalanceData = mock(StreamsRebalanceData.class);
        when(streamsRebalanceData.subtopologies()).thenReturn(Map.of());
        createRebalanceListenerWithRebalanceData(streamsRebalanceData);

        final Exception actualException = assertThrows(RuntimeException.class, () -> defaultStreamsRebalanceListener.onAllTasksLost());

        assertEquals(exception, actualException);
        verify(taskManager).handleLostAll();
        verify(streamsRebalanceData, never()).setReconciledAssignment(any());
    }
}
Repository: apache/kafka
File: release/git.py (Python)

Code before:
# 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.
#

"""
Auxiliary function to interact with git.
"""

import os
import subprocess

from runtime import repo_dir, execute, cmd

push_remote_name = os.environ.get("PUSH_REMOTE_NAME", "apache-github")


def __defaults(kwargs):
    if "cwd" not in kwargs:
        kwargs["cwd"] = repo_dir


def has_staged_changes(**kwargs):
    __defaults(kwargs)
    try:
        # 0: no staged changes; non-zero: there are staged changes
        execute("git diff --cached --exit-code --quiet", **kwargs)
        return False
    except subprocess.CalledProcessError:
        return True 


def has_unstaged_changes(**kwargs):
    __defaults(kwargs)
    try:
        # 0: no unstaged changes; non-zero: there are unstaged changes
        execute("git diff --exit-code --quiet", **kwargs)
        return False
    except subprocess.CalledProcessError:
        return True 


def fetch_tags(remote=push_remote_name, **kwargs):
    __defaults(kwargs)
    cmd(f"Fetching tags from {remote}", f"git fetch --tags {remote}", **kwargs)


def tags(**kwargs):
    __defaults(kwargs)
    return execute("git tag", **kwargs).split()


Review comment (bug):
This is wrong. 
non-zero(has diff) will raise exception since function `execute` use `subprocess.check_output`

Code after:
# 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.
#

"""
Auxiliary function to interact with git.
"""

import os

from runtime import repo_dir, execute, cmd

push_remote_name = os.environ.get("PUSH_REMOTE_NAME", "apache-github")


def __defaults(kwargs):
    if "cwd" not in kwargs:
        kwargs["cwd"] = repo_dir


def ensure_no_staged_changes(**kwargs):
    __defaults(kwargs)
    execute("git diff --cached --exit-code --quiet", **kwargs)
    return True


def ensure_no_unstaged_changes(**kwargs):
    __defaults(kwargs)
    execute("git diff --exit-code --quiet", **kwargs)
    return True


def fetch_tags(remote=push_remote_name, **kwargs):
    __defaults(kwargs)
    cmd(f"Fetching tags from {remote}", f"git fetch --tags {remote}", **kwargs)


def tags(**kwargs):
    __defaults(kwargs)
    return execute("git tag", **kwargs).split()


def tag_exists(tag, **kwargs):
    __defaults(kwargs)
    return tag in tags(**kwargs)


def delete_tag(tag, **kwargs):
    __defaults(kwargs)
    if tag_exists(tag, **kwargs):
Repository: apache/kafka
File: server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java (Java)

Code before:
    /**
     * The runnable that our thread executes.
     */
    private final EventHandler eventHandler;

    /**
     * The queue thread.
     */
    private final Thread eventHandlerThread;

    /**
     * True if the event queue is shutting down. Protected by the lock.
     */
    private boolean shuttingDown;

    /**
     * True if the event handler thread was interrupted. Protected by the lock.
     */
    private boolean interrupted;

    /**
     * Optional callback for queue idle time tracking.
     */
    private final BiConsumer<Long, Long> idleTimeCallback;


    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix
    ) {
        this(time, logContext, threadNamePrefix, VoidEvent.INSTANCE, (__, ___) -> { });
    }

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent
    ) {
        this(time, logContext, threadNamePrefix, cleanupEvent, (__, ___) -> { });
    }

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent,
        BiConsumer<Long, Long> idleTimeCallback
    ) {
        this.time = time;

Review comment (suggestion):
I recommend documenting the two parameters. E.g. what is the first Long vs the second Long to the BiConsumer.

Code after:
    /**
     * The runnable that our thread executes.
     */
    private final EventHandler eventHandler;

    /**
     * The queue thread.
     */
    private final Thread eventHandlerThread;

    /**
     * True if the event queue is shutting down. Protected by the lock.
     */
    private boolean shuttingDown;

    /**
     * True if the event handler thread was interrupted. Protected by the lock.
     */
    private boolean interrupted;

    /**
     * Optional callback for tracking queue idle time. The BiConsumer accepts two parameters:
     * the first Long is the idle duration in milliseconds, and the second Long is the current
     * time in milliseconds when the idle period ended. Both values are captured at the same
     * moment to ensure timing consistency for metric calculations.
     */
    private final BiConsumer<Long, Long> idleTimeCallback;


    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix
    ) {
        this(time, logContext, threadNamePrefix, VoidEvent.INSTANCE, (__, ___) -> { });
    }

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent
    ) {
        this(time, logContext, threadNamePrefix, cleanupEvent, (__, ___) -> { });
    }

    public KafkaEventQueue(
        Time time,
        LogContext logContext,
        String threadNamePrefix,
        Event cleanupEvent,
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java (Java)

Code before:
    }

    /** True if the current stream thread requested a shutdown of all Streams clients belonging to the same application. */
    public boolean shutdownRequested() {
        return shutdownRequested.get();
    }

    /** Updated whenever the status of the streams group is updated. */
    public void setStatuses(final List<StreamsGroupHeartbeatResponseData.Status> s) {
        statuses.set(s);
    }

    /** For communicating the current status of the group to the stream thread */
    public List<StreamsGroupHeartbeatResponseData.Status> statuses() {
        return statuses.get();
    }

    /** Updated whenever a heartbeat response is received from the broker. */
    public void setHeartbeatIntervalMs(final int heartbeatIntervalMs) {
        this.heartbeatIntervalMs.set(heartbeatIntervalMs);
    }

    /** Returns the heartbeat interval in milliseconds, or -1 if not yet set. */
    public int getHeartbeatIntervalMs() {
        return heartbeatIntervalMs.get();
    }

}


Review comment (suggestion):
The getter method's documentation states it returns -1 if not yet set, but the field is initialized to -1 via `AtomicInteger(-1)` on line 333. Consider documenting this initialization in the field-level comment or the setter to clarify when/how the value transitions from -1 to a valid heartbeat interval.

Code after:
    }

    /** True if the current stream thread requested a shutdown of all Streams clients belonging to the same application. */
    public boolean shutdownRequested() {
        return shutdownRequested.get();
    }

    /** Updated whenever the status of the streams group is updated. */
    public void setStatuses(final List<StreamsGroupHeartbeatResponseData.Status> s) {
        statuses.set(s);
    }

    /** For communicating the current status of the group to the stream thread */
    public List<StreamsGroupHeartbeatResponseData.Status> statuses() {
        return statuses.get();
    }

    /** Updated whenever a heartbeat response is received from the broker. */
    public void setHeartbeatIntervalMs(final int heartbeatIntervalMs) {
        this.heartbeatIntervalMs.set(heartbeatIntervalMs);
    }

    /** Returns the heartbeat interval in milliseconds, or -1 if not yet set. */
    public int heartbeatIntervalMs() {
        return heartbeatIntervalMs.get();
    }

}
Repository: apache/kafka
File: server/src/main/java/org/apache/kafka/network/Session.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.network;

import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.utils.Sanitizer;

import java.net.InetAddress;

public record Session(KafkaPrincipal principal, InetAddress clientAddress) {
    public String sanitizedUser() {
        return Sanitizer.sanitize(principal.getName());
    }
}


Review comment (performance):
On second thought, this might introduces a performance regression. Would you mind initializing it in the constructor 

Code after:
/*
 * 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.network;

import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.utils.Sanitizer;

import java.net.InetAddress;

public class Session {
    public final KafkaPrincipal principal;
    public final InetAddress clientAddress;
    public final String sanitizedUser;

    public Session(KafkaPrincipal principal, InetAddress clientAddress) {
        this.principal = principal;
        this.clientAddress = clientAddress;
        this.sanitizedUser = Sanitizer.sanitize(principal.getName());
    }
}
Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java (Java)

Code before:
    /**
     * The serializer used to serialize records.
     */
    private final Serializer<U> serializer;

    /**
     * The compression codec used when writing records.
     */
    private final Compression compression;

    /**
     * The duration in milliseconds that the coordinator will wait for writes to
     * accumulate before flushing them to disk.
     */
    private final int appendLingerMs;

    /**
     * The executor service used by the coordinator runtime to schedule
     * asynchronous tasks.
     */
    private final ExecutorService executorService;

    /**
     * The maximum buffer size that the coordinator can cache.
     */
    private final Supplier<Integer> appendMaxBufferSizeSupplied;

    /**
     * Atomic boolean indicating whether the runtime is running.
     */
    private final AtomicBoolean isRunning = new AtomicBoolean(true);

    /**
     * The latest known metadata image.
     */
    private volatile CoordinatorMetadataImage metadataImage = CoordinatorMetadataImage.EMPTY;

    /**
     * Constructor.
     *
     * @param logPrefix                         The log prefix.
     * @param logContext                        The log context.
     * @param processor                         The event processor.
     * @param partitionWriter                   The partition writer.
     * @param loader                            The coordinator loader.
     * @param coordinatorShardBuilderSupplier   The coordinator builder.
     * @param time                              The system time.
     * @param timer                             The system timer.
     * @param defaultWriteTimeout               The write operation timeout.
     * @param runtimeMetrics                    The runtime metrics.
     * @param coordinatorMetrics                The coordinator metrics.

Review comment (suggestion):
typo?
```suggestion
    private final Supplier<Integer> appendMaxBufferSizeSupplier;
```

Code after:
    static final int INITIAL_BUFFER_SIZE = 512 * 1024;

    /**
     * The log prefix.
     */
    private final String logPrefix;

    /**
     * The logger.
     */
    private final Logger log;

    /**
     * The system time.
     */
    private final Time time;

    /**
     * The system timer.
     */
    private final Timer timer;

    /**
     * The write operation timeout
     */
    private final Duration defaultWriteTimeout;

    /**
     * The coordinators keyed by topic partition.
     */
    private final ConcurrentHashMap<TopicPartition, CoordinatorContext> coordinators;

    /**
     * The event processor used by the runtime.
     */
    private final CoordinatorEventProcessor processor;

    /**
     * The partition writer used by the runtime to persist records.
     */
    private final PartitionWriter partitionWriter;

    /**
     * The coordinator loaded used by the runtime.
     */
    private final CoordinatorLoader<U> loader;

    /**
     * The coordinator state machine builder used by the runtime
     * to instantiate a coordinator.
     */
Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java (Java)

Code before:

    /**
     * Metric to count the number of partitions in Loading state.
     */
    private final MetricName numPartitionsLoading;
    private final AtomicLong numPartitionsLoadingCounter = new AtomicLong(0);

    /**
     * Metric to count the number of partitions in Active state.
     */
    private final MetricName numPartitionsActive;
    private final AtomicLong numPartitionsActiveCounter = new AtomicLong(0);

    /**
     * Metric to count the number of partitions in Failed state.
     */
    private final MetricName numPartitionsFailed;
    private final AtomicLong numPartitionsFailedCounter = new AtomicLong(0);

    /**
     * Metric to count the size of the processor queue.
     */
    private final MetricName eventQueueSize;

    /**
     * Metric to count the size of the cached buffer.
     */
    private final MetricName bufferCacheSize;

    /**
     * Metric to count the number of over-sized append buffers that were discarded.
     */
    private final MetricName bufferCacheDiscardCount;
    private final AtomicLong bufferCacheDiscardCounter = new AtomicLong(0);

    /**
     * The Kafka metrics registry.
     */
    private final Metrics metrics;

    /**
     * The partition load sensor.
     */
    private final Sensor partitionLoadSensor;

    /**
     * The thread idle sensor.
     */
    private final Sensor threadIdleSensor;

    /**

Review comment (performance):
nit:
```suggestion
     * Metric to count the size of the cached buffers.
```

Code after:

    /**
     * Metric to count the number of partitions in Loading state.
     */
    private final MetricName numPartitionsLoading;
    private final AtomicLong numPartitionsLoadingCounter = new AtomicLong(0);

    /**
     * Metric to count the number of partitions in Active state.
     */
    private final MetricName numPartitionsActive;
    private final AtomicLong numPartitionsActiveCounter = new AtomicLong(0);

    /**
     * Metric to count the number of partitions in Failed state.
     */
    private final MetricName numPartitionsFailed;
    private final AtomicLong numPartitionsFailedCounter = new AtomicLong(0);

    /**
     * Metric to count the size of the processor queue.
     */
    private final MetricName eventQueueSize;

    /**
     * Metric to count the size of the cached buffers.
     */
    private final MetricName bufferCacheSize;

    /**
     * Metric to count the number of over-sized append buffers that were discarded.
     */
    private final MetricName bufferCacheDiscardCount;
    private final AtomicLong bufferCacheDiscardCounter = new AtomicLong(0);

    /**
     * The Kafka metrics registry.
     */
    private final Metrics metrics;

    /**
     * The partition load sensor.
     */
    private final Sensor partitionLoadSensor;

    /**
     * The thread idle sensor.
     */
    private final Sensor threadIdleSensor;

    /**
Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java (Java)

Code before:

        private final String name;

        private final String lowerCaseName;

        ConsumerGroupState(String name) {
            this.name = name;
            this.lowerCaseName = name.toLowerCase(Locale.ROOT);
        }

        @Override
        public String toString() {
            return name;
        }

        public String toLowerCaseString() {
            return lowerCaseName;
        }
    }

    private final Logger log;

    /**
     * The group state.
     */
    private final TimelineObject<ConsumerGroupState> state;

    /**
     * The static group members.
     */
    private final TimelineHashMap<String, String> staticMembers;

    /**
     * The number of members supporting each server assignor name.
     */
    private final TimelineHashMap<String, Integer> serverAssignors;

    /**
     * The number of members that use the classic protocol.
     */
    private final TimelineInteger numClassicProtocolMembers;

    /**
     * Map of protocol names to the number of members that use classic protocol and support them.
     */
    private final TimelineHashMap<String, Integer> classicProtocolMembersSupportedProtocols;

    /**
     * The current partition epoch maps each topic-partitions to their current epoch where
     * the epoch is the epoch of their owners. When a member revokes a partition, it removes
     * its epochs from this map. When a member gets a partition, it adds its epochs to this map.

Review comment (suggestion):
```suggestion
    /**
     * The logger.
     */
    private final Logger log;
```

Code after:
        STABLE("Stable"),
        DEAD("Dead");

        private final String name;

        private final String lowerCaseName;

        ConsumerGroupState(String name) {
            this.name = name;
            this.lowerCaseName = name.toLowerCase(Locale.ROOT);
        }

        @Override
        public String toString() {
            return name;
        }

        public String toLowerCaseString() {
            return lowerCaseName;
        }
    }

    /**
     * The logger.
     */
    private final Logger log;

    /**
     * The group state.
     */
    private final TimelineObject<ConsumerGroupState> state;

    /**
     * The static group members.
     */
    private final TimelineHashMap<String, String> staticMembers;

    /**
     * The number of members supporting each server assignor name.
     */
    private final TimelineHashMap<String, Integer> serverAssignors;

    /**
     * The number of members that use the classic protocol.
     */
    private final TimelineInteger numClassicProtocolMembers;

    /**
     * Map of protocol names to the number of members that use classic protocol and support them.
     */
    private final TimelineHashMap<String, Integer> classicProtocolMembersSupportedProtocols;