Merge remote-tracking branch 'apache-kafka/trunk' into ccs-kafka/master

[omkreddy]

    Conflicts:
            gradle.properties - minor conflict due to ccs version

[Copilot]

Pull Request Overview

This is a merge commit from the Apache Kafka trunk repository that incorporates upstream changes with a minor conflict resolution in gradle.properties due to CCS versioning. The merge updates various components including dependency versions, test improvements, metrics infrastructure changes, and GitHub Actions workflow updates.

Key changes include:

• Updated dependency versions (Scala, Mockito, Zinc) and GitHub Actions versions
• Refactored metrics infrastructure to use explicit package/class names instead of reflection
• Enhanced test coverage with new test methods and improved assertions
• Fixed logging level in StateChangeLogger from info to trace

Reviewed Changes

Copilot reviewed 39 out of 42 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
gradle/dependencies.gradle	Updates Scala, Mockito, and Zinc dependency versions
gradle.properties	Updates Scala version with CCS version conflict resolution
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaMetricsGroup.java	Removes deprecated constructor that used reflection
Multiple core Scala files	Replaces reflection-based metrics with explicit package/class names
server-common/src/main/java/org/apache/kafka/logger/StateChangeLogger.java	Fixes logging level from info to trace
Various test files	Adds new tests and improves existing test assertions
GitHub workflow files	Updates action versions for security and compatibility

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[Copilot]

[nitpick] The test logic has changed from checking for empty output to checking that output doesn't contain the topic name. This is a less strict assertion that could miss other unexpected output. Consider adding an additional check to ensure the output is actually empty or only contains expected content.

Suggested change

	String.format("--under-replicated-partitions shouldn't contain '%s': '%s'", testTopicNameLogLine, underReplicatedOutput));
	String.format("--under-replicated-partitions shouldn't contain '%s': '%s'", testTopicNameLogLine, underReplicatedOutput));
	assertTrue(underReplicatedOutput.trim().isEmpty(),
	String.format("--under-replicated-partitions output should be empty, but was: '%s'", underReplicatedOutput));

[Copilot]

These new methods expose internal mutable collections directly without defensive copying. This violates encapsulation and could allow external modification of internal state. Consider returning unmodifiable views or defensive copies.

Suggested change

	return pendingActiveTasksToCreate;
	}
	Map<TaskId, Set<TopicPartition>> pendingStandbyTasksToCreate() {
	return pendingStandbyTasksToCreate;
	Map<TaskId, Set<TopicPartition>> unmodifiableMap = new HashMap<>();
	for (Map.Entry<TaskId, Set<TopicPartition>> entry : pendingActiveTasksToCreate.entrySet()) {
	unmodifiableMap.put(entry.getKey(), Collections.unmodifiableSet(entry.getValue()));
	}
	return Collections.unmodifiableMap(unmodifiableMap);
	}
	Map<TaskId, Set<TopicPartition>> pendingStandbyTasksToCreate() {
	Map<TaskId, Set<TopicPartition>> unmodifiableMap = new HashMap<>();
	for (Map.Entry<TaskId, Set<TopicPartition>> entry : pendingStandbyTasksToCreate.entrySet()) {
	unmodifiableMap.put(entry.getKey(), Collections.unmodifiableSet(entry.getValue()));
	}
	return Collections.unmodifiableMap(unmodifiableMap);

[Copilot]

The variables prevLeaderReplicaIdOpt and prevLeaderEpoch capture the current values before they are updated, but these values will be the same as the new values if this is not a new leader epoch. The logging message will be incorrect in cases where the leader hasn't changed.

Suggested change

	val isNewLeaderEpoch = partitionRegistration.leaderEpoch > leaderEpoch
	// The leader should be updated before updateAssignmentAndIsr where we clear the ISR. Or it is possible to meet
	// the under min isr condition during the makeFollower process and emits the wrong metric.
	val prevLeaderReplicaIdOpt = leaderReplicaIdOpt
	val prevLeaderEpoch = leaderEpoch
	val prevLeaderReplicaIdOpt = leaderReplicaIdOpt
	val prevLeaderEpoch = leaderEpoch
	val isNewLeaderEpoch = partitionRegistration.leaderEpoch > leaderEpoch
	// The leader should be updated before updateAssignmentAndIsr where we clear the ISR. Or it is possible to meet
	// the under min isr condition during the makeFollower process and emits the wrong metric.