jsancio commented on code in PR #21221:
URL: https://github.com/apache/kafka/pull/21221#discussion_r2683939084
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1788,21 +1808,27 @@ void
maybeTriggerUncleanLeaderElectionForLeaderlessPartitions(
Iterator<TopicIdPartition> iterator =
brokersToIsrs.partitionsWithNoLeader();
while (iterator.hasNext() && records.size() < maxElections) {
TopicIdPartition topicIdPartition = iterator.next();
+ int partitionId = topicIdPartition.partitionId();
TopicControlInfo topic = topics.get(topicIdPartition.topicId());
+
+ PartitionRegistration partition = topic.parts.get(partitionId);
+ String partitionInfo = String.format("(isr: %s, replicaSet: %s,
partitionEpoch: %d, leaderEpoch: %d)",
+ Arrays.toString(partition.isr),
Arrays.toString(partition.replicas), partition.partitionEpoch,
partition.leaderEpoch);
Review Comment:
Same comment here. This can be turned into a function as it will only be
called once given the control flow structure.
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1247,40 +1260,48 @@ private Errors validateAlterPartitionData(
return UNKNOWN_TOPIC_OR_PARTITION;
}
+ String partitionChangeInfo = String.format("Proposed ISR was %s and
current ISR is %s. " +
+ "Current replica set is %s. Proposed partitionEpoch was %d and
current partitionEpoch is %d. " +
+ "Proposed leaderEpoch was %d and current leaderEpoch is %d.",
Review Comment:
Senders of the ALTER_PARTITION request do not propose leader epoch or
partition epoch. The active controller instead check that this value match to
make sure that the changes are based on the same state. This is how Kafka
implement optimistic locking for partition changes.
Every usage of `partitionChangeInfo` is followed by a `return`. So you can
move this to a helper as it will only be called once.
##########
metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java:
##########
@@ -1141,10 +1147,18 @@ ControllerResult<AlterPartitionResponseData>
alterPartition(
if (record.isPresent()) {
records.add(record.get());
PartitionChangeRecord change = (PartitionChangeRecord)
record.get().message();
+ String priorIsr = Arrays.toString(partition.isr);
+ String priorReplicaSet =
Arrays.toString(partition.replicas);
+ int priorPartitionEpoch = partition.partitionEpoch;
+ int priorLeaderEpoch = partition.leaderEpoch;
partition = partition.merge(change);
+ String partitionChangeInfo = String.format("isr: %s -> %s,
replicaSet: %s -> %s, " +
+ "partitionEpoch: %d -> %d, leaderEpoch: %d -> %d",
+ priorIsr, Arrays.toString(partition.isr),
priorReplicaSet, Arrays.toString(partition.replicas),
+ priorPartitionEpoch, partition.partitionEpoch,
priorLeaderEpoch, partition.leaderEpoch);
Review Comment:
All of this work should only be done if debug logging is enabled. You can
take advantage of the fact that PartitionRegistration are immutable and that
PartitionRegistration#merge is side effect free to make sure that all of this
formatting is only performed if debug logging is enabled.
Can you move this to a private static method? This a lot of code to generate
a string which distracts from the core functionality of the method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
