You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/04/19 21:50:01 UTC
[GitHub] [kafka] cmccabe opened a new pull request #10564: MINOR: clean up some replication code
cmccabe opened a new pull request #10564:
URL: https://github.com/apache/kafka/pull/10564
Centralize leader and ISR changes in generateLeaderAndIsrUpdates.
Consolidate handleNodeDeactivated and handleNodeActivated into this
function.
Rename BrokersToIsrs#noLeaderIterator to BrokersToIsrs#partitionsWithNoLeader.
Create BrokersToIsrs#partitionsLedByBroker, BrokersToIsrs#partitionsWithBrokerInIsr
In ReplicationControlManagerTest, createTestTopic should be a member
function of ReplicationControlTestContext. It should invoke
ReplicationControlTestContext#replay so that records are applied to all
parts of the test context.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618132526
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -356,7 +369,8 @@ public void replay(PartitionChangeRecord record) {
brokersToIsrs.update(record.topicId(), record.partitionId(),
prevPartitionInfo.isr, newPartitionInfo.isr, prevPartitionInfo.leader,
newPartitionInfo.leader);
- log.debug("Applied ISR change record: {}", record.toString());
+ String topicPart = topicInfo.name + "-" + record.partitionId();
Review comment:
sorry. added.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617297824
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -321,22 +339,17 @@ public void replay(PartitionRecord record) {
}
PartitionControlInfo newPartInfo = new PartitionControlInfo(record);
PartitionControlInfo prevPartInfo = topicInfo.parts.get(record.partitionId());
+ String topicPart = topicInfo.name + "-" + record.partitionId();
if (prevPartInfo == null) {
- log.info("Created partition {}:{} with {}.", record.topicId(),
- record.partitionId(), newPartInfo.toString());
+ log.info("Created partition {} with {}.", topicPart, newPartInfo.toString());
Review comment:
Is it intentional to use "name" to replace "id"?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1053,62 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemoveFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemoveFromIsr);
+ int newLeader = Replicas.contains(newIsr, partition.leader) ? partition.leader :
+ bestLeader(partition.replicas, newIsr, false);
+ boolean unclean = newLeader != NO_LEADER && !Replicas.contains(newIsr, newLeader);
Review comment:
Could we reuse `wasCleanlyDerivedFrom` to get "unclean"?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -907,13 +843,13 @@ ApiError electLeader(String topic, int partitionId, boolean unclean,
}
PartitionChangeRecord record = new PartitionChangeRecord().
setPartitionId(partitionId).
- setTopicId(topicId);
- if (unclean && !Replicas.contains(partitionInfo.isr, newLeader)) {
- // If the election was unclean, we may have to forcibly add the replica to
- // the ISR. This can result in data loss!
+ setTopicId(topicId).
+ setLeader(newLeader);
+ if (!Replicas.contains(partitionInfo.isr, newLeader)) {
+ // If the election was unclean, we have to forcibly set the ISR to just the
+ // new leader. This can result in data loss!
Review comment:
redundant "space"
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1053,62 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemoveFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemoveFromIsr);
+ int newLeader = Replicas.contains(newIsr, partition.leader) ? partition.leader :
+ bestLeader(partition.replicas, newIsr, false);
+ boolean unclean = newLeader != NO_LEADER && !Replicas.contains(newIsr, newLeader);
+ if (unclean) {
+ // After an unclean leader election, the ISR is reset to just the new leader.
+ newIsr = new int[] {newLeader};
+ } else if (newIsr.length == 0) {
+ // We never want to shrink the ISR to size 0.
Review comment:
The `newLeader` is `NO_LEADER` (due to empty `newIsr`) so we could set `NO_LEADER` to `record`. Does it make sense?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618135556
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
Review comment:
In the case of an unclean leader election, I suppose we do need to consider brokerToRemove, since we'll want to explicitly exclude it. Good catch.
In general the code will work if both brokerToRemove and brokerToAdd are both set, although we don't plan to do this.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617507396
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1053,62 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemoveFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemoveFromIsr);
+ int newLeader = Replicas.contains(newIsr, partition.leader) ? partition.leader :
+ bestLeader(partition.replicas, newIsr, false);
+ boolean unclean = newLeader != NO_LEADER && !Replicas.contains(newIsr, newLeader);
+ if (unclean) {
+ // After an unclean leader election, the ISR is reset to just the new leader.
+ newIsr = new int[] {newLeader};
+ } else if (newIsr.length == 0) {
+ // We never want to shrink the ISR to size 0.
Review comment:
I agree that if `newIsr.length` is 0 on line 1080, then `newLeader` must be `NO_LEADER`. However, it's not necessary to add any special logic for this since `bestLeader` already handles this case.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617093936
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1053,61 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemoveFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemoveFromIsr);
+ int newLeader = bestLeader(partition.replicas, newIsr, false);
Review comment:
One side effect of doing this is that it can force the leader to change unnecessarily. For example, the current leader could still be in newIsr, but may not be the first in replica list. bestLeader() will cause the leader to change in this case.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618133367
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -172,47 +174,54 @@ String diff(PartitionControlInfo prev) {
StringBuilder builder = new StringBuilder();
String prefix = "";
if (!Arrays.equals(replicas, prev.replicas)) {
- builder.append(prefix).append("oldReplicas=").append(Arrays.toString(prev.replicas));
+ builder.append(prefix).append("replicas: ").
+ append(Arrays.toString(prev.replicas)).
+ append(" -> ").append(Arrays.toString(replicas));
prefix = ", ";
- builder.append(prefix).append("newReplicas=").append(Arrays.toString(replicas));
}
if (!Arrays.equals(isr, prev.isr)) {
- builder.append(prefix).append("oldIsr=").append(Arrays.toString(prev.isr));
+ builder.append(prefix).append("isr: ").
+ append(Arrays.toString(prev.isr)).
+ append(" -> ").append(Arrays.toString(isr));
prefix = ", ";
- builder.append(prefix).append("newIsr=").append(Arrays.toString(isr));
}
if (!Arrays.equals(removingReplicas, prev.removingReplicas)) {
- builder.append(prefix).append("oldRemovingReplicas=").
- append(Arrays.toString(prev.removingReplicas));
+ builder.append(prefix).append("removingReplicas: ").
+ append(Arrays.toString(prev.removingReplicas)).
+ append(" -> ").append(Arrays.toString(removingReplicas));
prefix = ", ";
- builder.append(prefix).append("newRemovingReplicas=").
- append(Arrays.toString(removingReplicas));
}
if (!Arrays.equals(addingReplicas, prev.addingReplicas)) {
- builder.append(prefix).append("oldAddingReplicas=").
- append(Arrays.toString(prev.addingReplicas));
+ builder.append(prefix).append("addingReplicas: ").
+ append(Arrays.toString(prev.addingReplicas)).
+ append(" -> ").append(Arrays.toString(addingReplicas));
prefix = ", ";
- builder.append(prefix).append("newAddingReplicas=").
- append(Arrays.toString(addingReplicas));
}
if (leader != prev.leader) {
- builder.append(prefix).append("oldLeader=").append(prev.leader);
+ builder.append(prefix).append("leader: ").
+ append(prev.leader).append(" -> ").append(leader);
prefix = ", ";
- builder.append(prefix).append("newLeader=").append(leader);
}
if (leaderEpoch != prev.leaderEpoch) {
- builder.append(prefix).append("oldLeaderEpoch=").append(prev.leaderEpoch);
+ builder.append(prefix).append("leaderEpoch: ").
+ append(prev.leaderEpoch).append(" -> ").append(leaderEpoch);
prefix = ", ";
- builder.append(prefix).append("newLeaderEpoch=").append(leaderEpoch);
}
if (partitionEpoch != prev.partitionEpoch) {
- builder.append(prefix).append("oldPartitionEpoch=").append(prev.partitionEpoch);
- prefix = ", ";
- builder.append(prefix).append("newPartitionEpoch=").append(partitionEpoch);
+ builder.append(prefix).append("partitionEpoch: ").
+ append(prev.partitionEpoch).append(" -> ").append(partitionEpoch);
}
return builder.toString();
}
+ void maybeLogPartitionChange(Logger log, String description, PartitionControlInfo prev) {
+ if (!electionWasClean(leader, prev.isr)) {
+ log.info("UNCLEAN partition change for {}: {}", description, diff(prev));
+ } else if (log.isDebugEnabled()) {
+ log.debug("partition change for {}: {}", description, diff(prev));
Review comment:
I don't think this will be practical as the number of partitions grows. A 10-node cluster with a million partitions could have tens of thousands of partition changes when a node goes away or comes back.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618761439
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
Review comment:
What about this place? It seems that Instead of relying upon NO_LEADER, it might be clearer to make brokerToRemove and brokerToAdd optional int?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() +
+ " existed in isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemove);
+ int newLeader;
+ if (isGoodLeader(newIsr, partition.leader)) {
+ // If the current leader is good, don't change.
+ newLeader = partition.leader;
+ } else {
+ // Choose a new leader.
+ boolean uncleanOk = configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name);
+ newLeader = bestLeader(partition.replicas, newIsr, uncleanOk, isAcceptableLeader);
Review comment:
Yes, it probably doesn't make a big difference and it happens rarely. So, we could just keep the logic in this PR.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r616258845
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1039,77 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ boolean excludeCurrentLeaderFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = partition.isr;
+ if (excludeCurrentLeaderFromIsr) {
+ newIsr = Replicas.copyWithout(partition.isr, partition.leader);
+ }
+ int newLeader = bestLeader(partition.replicas, newIsr, false);
+ boolean unclean = newLeader != NO_LEADER && !Replicas.contains(newIsr, newLeader);
+ if (unclean) {
+ // After an unclean leader election, the ISR is reset to just the new leader.
+ newIsr = new int[] {newLeader};
+ } else if (newIsr.length == 0) {
+ // We never want to shrink the ISR to size 0.
+ newIsr = partition.isr;
+ }
+ PartitionChangeRecord record = new PartitionChangeRecord();
+ if (newLeader != partition.leader) {
+ record.setLeader(newLeader);
+ }
+ if (!Arrays.equals(newIsr, partition.isr)) {
+ record.setIsr(Replicas.toList(newIsr));
+ }
+ if (record.leader() != NO_LEADER_CHANGE || record.isr() != null) {
+ record.setPartitionId(topicIdPart.partitionId()).setTopicId(topic.id);
+ if (unclean) {
+ log.info("{}: UNCLEANLY {}", context,
+ leaderAndIsrUpdateLogMessage(topicIdPart, partition, record));
+ } else if (log.isDebugEnabled()) {
+ log.debug("{}: {}", context,
+ leaderAndIsrUpdateLogMessage(topicIdPart, partition, record));
+ }
+ records.add(new ApiMessageAndVersion(record, (short) 0));
+ }
+ }
+ }
+
+ // VisibleForTesting
+ String leaderAndIsrUpdateLogMessage(TopicIdPartition topicIdPart,
Review comment:
Ideally, we want to log this when the record is reflected through replay in the controller. That's also when we could log the new leaderEpoch.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618138346
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() +
+ " existed in isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemove);
+ int newLeader;
+ if (isGoodLeader(newIsr, partition.leader)) {
+ // If the current leader is good, don't change.
+ newLeader = partition.leader;
+ } else {
+ // Choose a new leader.
+ boolean uncleanOk = configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name);
+ newLeader = bestLeader(partition.replicas, newIsr, uncleanOk, isAcceptableLeader);
Review comment:
That will block the controlled shutdown from finishing in many cases. Is the intention to allow for a clean leader election later on if the other replicas catch up?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe merged pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe merged pull request #10564:
URL: https://github.com/apache/kafka/pull/10564
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617773890
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -321,22 +339,17 @@ public void replay(PartitionRecord record) {
}
PartitionControlInfo newPartInfo = new PartitionControlInfo(record);
PartitionControlInfo prevPartInfo = topicInfo.parts.get(record.partitionId());
+ String topicPart = topicInfo.name + "-" + record.partitionId();
Review comment:
It's probably useful to log the topicId too?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -356,7 +369,8 @@ public void replay(PartitionChangeRecord record) {
brokersToIsrs.update(record.topicId(), record.partitionId(),
prevPartitionInfo.isr, newPartitionInfo.isr, prevPartitionInfo.leader,
newPartitionInfo.leader);
- log.debug("Applied ISR change record: {}", record.toString());
+ String topicPart = topicInfo.name + "-" + record.partitionId();
Review comment:
It's probably useful to log the topicId too?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617858509
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -172,47 +174,54 @@ String diff(PartitionControlInfo prev) {
StringBuilder builder = new StringBuilder();
String prefix = "";
if (!Arrays.equals(replicas, prev.replicas)) {
- builder.append(prefix).append("oldReplicas=").append(Arrays.toString(prev.replicas));
+ builder.append(prefix).append("replicas: ").
+ append(Arrays.toString(prev.replicas)).
+ append(" -> ").append(Arrays.toString(replicas));
prefix = ", ";
- builder.append(prefix).append("newReplicas=").append(Arrays.toString(replicas));
}
if (!Arrays.equals(isr, prev.isr)) {
- builder.append(prefix).append("oldIsr=").append(Arrays.toString(prev.isr));
+ builder.append(prefix).append("isr: ").
+ append(Arrays.toString(prev.isr)).
+ append(" -> ").append(Arrays.toString(isr));
prefix = ", ";
- builder.append(prefix).append("newIsr=").append(Arrays.toString(isr));
}
if (!Arrays.equals(removingReplicas, prev.removingReplicas)) {
- builder.append(prefix).append("oldRemovingReplicas=").
- append(Arrays.toString(prev.removingReplicas));
+ builder.append(prefix).append("removingReplicas: ").
+ append(Arrays.toString(prev.removingReplicas)).
+ append(" -> ").append(Arrays.toString(removingReplicas));
prefix = ", ";
- builder.append(prefix).append("newRemovingReplicas=").
- append(Arrays.toString(removingReplicas));
}
if (!Arrays.equals(addingReplicas, prev.addingReplicas)) {
- builder.append(prefix).append("oldAddingReplicas=").
- append(Arrays.toString(prev.addingReplicas));
+ builder.append(prefix).append("addingReplicas: ").
+ append(Arrays.toString(prev.addingReplicas)).
+ append(" -> ").append(Arrays.toString(addingReplicas));
prefix = ", ";
- builder.append(prefix).append("newAddingReplicas=").
- append(Arrays.toString(addingReplicas));
}
if (leader != prev.leader) {
- builder.append(prefix).append("oldLeader=").append(prev.leader);
+ builder.append(prefix).append("leader: ").
+ append(prev.leader).append(" -> ").append(leader);
prefix = ", ";
- builder.append(prefix).append("newLeader=").append(leader);
}
if (leaderEpoch != prev.leaderEpoch) {
- builder.append(prefix).append("oldLeaderEpoch=").append(prev.leaderEpoch);
+ builder.append(prefix).append("leaderEpoch: ").
+ append(prev.leaderEpoch).append(" -> ").append(leaderEpoch);
prefix = ", ";
- builder.append(prefix).append("newLeaderEpoch=").append(leaderEpoch);
}
if (partitionEpoch != prev.partitionEpoch) {
- builder.append(prefix).append("oldPartitionEpoch=").append(prev.partitionEpoch);
- prefix = ", ";
- builder.append(prefix).append("newPartitionEpoch=").append(partitionEpoch);
+ builder.append(prefix).append("partitionEpoch: ").
+ append(prev.partitionEpoch).append(" -> ").append(partitionEpoch);
}
return builder.toString();
}
+ void maybeLogPartitionChange(Logger log, String description, PartitionControlInfo prev) {
+ if (!electionWasClean(leader, prev.isr)) {
+ log.info("UNCLEAN partition change for {}: {}", description, diff(prev));
+ } else if (log.isDebugEnabled()) {
+ log.debug("partition change for {}: {}", description, diff(prev));
Review comment:
We use to log all leader and isr changes in info even for clean leader election.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -321,22 +336,18 @@ public void replay(PartitionRecord record) {
}
PartitionControlInfo newPartInfo = new PartitionControlInfo(record);
PartitionControlInfo prevPartInfo = topicInfo.parts.get(record.partitionId());
+ String description = topicInfo.name + "-" + record.partitionId() +
+ " with ID " + record.topicId();
Review comment:
with ID => with topic ID to make it clear?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -957,22 +890,27 @@ ApiError electLeader(String topic, int partitionId, boolean unclean,
return ControllerResult.of(records, reply);
}
- int bestLeader(int[] replicas, int[] isr, boolean unclean) {
+ static boolean isGoodLeader(int[] isr, int leader) {
+ return Replicas.contains(isr, leader);
+ }
+
+ static int bestLeader(int[] replicas, int[] isr, boolean uncleanOk,
Review comment:
Would it be better to return optional int here?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -957,22 +890,27 @@ ApiError electLeader(String topic, int partitionId, boolean unclean,
return ControllerResult.of(records, reply);
}
- int bestLeader(int[] replicas, int[] isr, boolean unclean) {
+ static boolean isGoodLeader(int[] isr, int leader) {
+ return Replicas.contains(isr, leader);
+ }
+
+ static int bestLeader(int[] replicas, int[] isr, boolean uncleanOk,
+ Function<Integer, Boolean> isAcceptableLeader) {
+ int bestUnclean = NO_LEADER;
for (int i = 0; i < replicas.length; i++) {
int replica = replicas[i];
- if (Replicas.contains(isr, replica)) {
- return replica;
- }
- }
- if (unclean) {
- for (int i = 0; i < replicas.length; i++) {
- int replica = replicas[i];
- if (clusterControl.unfenced(replica)) {
+ if (isAcceptableLeader.apply(replica)) {
+ if (bestUnclean == NO_LEADER) bestUnclean = replica;
+ if (Replicas.contains(isr, replica)) {
return replica;
}
}
}
- return NO_LEADER;
+ return uncleanOk ? bestUnclean : NO_LEADER;
+ }
+
+ static boolean electionWasClean(int newLeader, int[] prevIsr) {
Review comment:
Since we don't always pass in previous isr to prevIsr, perhaps we could name it more accurately?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -356,7 +369,8 @@ public void replay(PartitionChangeRecord record) {
brokersToIsrs.update(record.topicId(), record.partitionId(),
prevPartitionInfo.isr, newPartitionInfo.isr, prevPartitionInfo.leader,
newPartitionInfo.leader);
- log.debug("Applied ISR change record: {}", record.toString());
+ String topicPart = topicInfo.name + "-" + record.partitionId();
Review comment:
It seems that we haven't added the topicId yet.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() +
+ " existed in isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemove);
+ int newLeader;
+ if (isGoodLeader(newIsr, partition.leader)) {
+ // If the current leader is good, don't change.
+ newLeader = partition.leader;
+ } else {
+ // Choose a new leader.
+ boolean uncleanOk = configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name);
+ newLeader = bestLeader(partition.replicas, newIsr, uncleanOk, isAcceptableLeader);
Review comment:
In the case of controlled shutdown, currently, we ignore the unclean leader election flag and always to elect a new leader cleanly.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
Review comment:
Instead of relying upon NO_LEADER, could we make brokerToRemove and brokerToAdd optional int to make them clearer?
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
Review comment:
Do we need to consider brokerToRemove here? Also, do we expect at least one of brokerToRemove and brokerToAdd to be NO_LEADER? If so, should we assert that?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618133367
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -172,47 +174,54 @@ String diff(PartitionControlInfo prev) {
StringBuilder builder = new StringBuilder();
String prefix = "";
if (!Arrays.equals(replicas, prev.replicas)) {
- builder.append(prefix).append("oldReplicas=").append(Arrays.toString(prev.replicas));
+ builder.append(prefix).append("replicas: ").
+ append(Arrays.toString(prev.replicas)).
+ append(" -> ").append(Arrays.toString(replicas));
prefix = ", ";
- builder.append(prefix).append("newReplicas=").append(Arrays.toString(replicas));
}
if (!Arrays.equals(isr, prev.isr)) {
- builder.append(prefix).append("oldIsr=").append(Arrays.toString(prev.isr));
+ builder.append(prefix).append("isr: ").
+ append(Arrays.toString(prev.isr)).
+ append(" -> ").append(Arrays.toString(isr));
prefix = ", ";
- builder.append(prefix).append("newIsr=").append(Arrays.toString(isr));
}
if (!Arrays.equals(removingReplicas, prev.removingReplicas)) {
- builder.append(prefix).append("oldRemovingReplicas=").
- append(Arrays.toString(prev.removingReplicas));
+ builder.append(prefix).append("removingReplicas: ").
+ append(Arrays.toString(prev.removingReplicas)).
+ append(" -> ").append(Arrays.toString(removingReplicas));
prefix = ", ";
- builder.append(prefix).append("newRemovingReplicas=").
- append(Arrays.toString(removingReplicas));
}
if (!Arrays.equals(addingReplicas, prev.addingReplicas)) {
- builder.append(prefix).append("oldAddingReplicas=").
- append(Arrays.toString(prev.addingReplicas));
+ builder.append(prefix).append("addingReplicas: ").
+ append(Arrays.toString(prev.addingReplicas)).
+ append(" -> ").append(Arrays.toString(addingReplicas));
prefix = ", ";
- builder.append(prefix).append("newAddingReplicas=").
- append(Arrays.toString(addingReplicas));
}
if (leader != prev.leader) {
- builder.append(prefix).append("oldLeader=").append(prev.leader);
+ builder.append(prefix).append("leader: ").
+ append(prev.leader).append(" -> ").append(leader);
prefix = ", ";
- builder.append(prefix).append("newLeader=").append(leader);
}
if (leaderEpoch != prev.leaderEpoch) {
- builder.append(prefix).append("oldLeaderEpoch=").append(prev.leaderEpoch);
+ builder.append(prefix).append("leaderEpoch: ").
+ append(prev.leaderEpoch).append(" -> ").append(leaderEpoch);
prefix = ", ";
- builder.append(prefix).append("newLeaderEpoch=").append(leaderEpoch);
}
if (partitionEpoch != prev.partitionEpoch) {
- builder.append(prefix).append("oldPartitionEpoch=").append(prev.partitionEpoch);
- prefix = ", ";
- builder.append(prefix).append("newPartitionEpoch=").append(partitionEpoch);
+ builder.append(prefix).append("partitionEpoch: ").
+ append(prev.partitionEpoch).append(" -> ").append(partitionEpoch);
}
return builder.toString();
}
+ void maybeLogPartitionChange(Logger log, String description, PartitionControlInfo prev) {
+ if (!electionWasClean(leader, prev.isr)) {
+ log.info("UNCLEAN partition change for {}: {}", description, diff(prev));
+ } else if (log.isDebugEnabled()) {
+ log.debug("partition change for {}: {}", description, diff(prev));
Review comment:
I don't think this will be practical as the number of partitions grows. A 10-node cluster with a million partitions could have hundreds of thousands of partition changes when a node goes away or comes back.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618138592
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -957,22 +890,27 @@ ApiError electLeader(String topic, int partitionId, boolean unclean,
return ControllerResult.of(records, reply);
}
- int bestLeader(int[] replicas, int[] isr, boolean unclean) {
+ static boolean isGoodLeader(int[] isr, int leader) {
+ return Replicas.contains(isr, leader);
+ }
+
+ static int bestLeader(int[] replicas, int[] isr, boolean uncleanOk,
+ Function<Integer, Boolean> isAcceptableLeader) {
+ int bestUnclean = NO_LEADER;
for (int i = 0; i < replicas.length; i++) {
int replica = replicas[i];
- if (Replicas.contains(isr, replica)) {
- return replica;
- }
- }
- if (unclean) {
- for (int i = 0; i < replicas.length; i++) {
- int replica = replicas[i];
- if (clusterControl.unfenced(replica)) {
+ if (isAcceptableLeader.apply(replica)) {
+ if (bestUnclean == NO_LEADER) bestUnclean = replica;
+ if (Replicas.contains(isr, replica)) {
return replica;
}
}
}
- return NO_LEADER;
+ return uncleanOk ? bestUnclean : NO_LEADER;
+ }
+
+ static boolean electionWasClean(int newLeader, int[] prevIsr) {
Review comment:
I will just call it isr.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r623286027
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -356,7 +367,9 @@ public void replay(PartitionChangeRecord record) {
brokersToIsrs.update(record.topicId(), record.partitionId(),
prevPartitionInfo.isr, newPartitionInfo.isr, prevPartitionInfo.leader,
newPartitionInfo.leader);
- log.debug("Applied ISR change record: {}", record.toString());
+ String topicPart = topicInfo.name + "-" + record.partitionId() + " with topic ID " +
Review comment:
Filed https://issues.apache.org/jira/browse/KAFKA-12733 for this
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r621652737
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() +
+ " existed in isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemove);
+ int newLeader;
+ if (isGoodLeader(newIsr, partition.leader)) {
+ // If the current leader is good, don't change.
+ newLeader = partition.leader;
+ } else {
+ // Choose a new leader.
+ boolean uncleanOk = configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name);
+ newLeader = bestLeader(partition.replicas, newIsr, uncleanOk, isAcceptableLeader);
Review comment:
The reason that we want to remove the shutting down replicas from ISR is to optimize for latency. If we don't do that, when the broker actually shuts down, it can block the producer for replica.max.ms before the replica can be taken out of ISR. So, I think this optimization is still useful.
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -356,7 +367,9 @@ public void replay(PartitionChangeRecord record) {
brokersToIsrs.update(record.topicId(), record.partitionId(),
prevPartitionInfo.isr, newPartitionInfo.isr, prevPartitionInfo.leader,
newPartitionInfo.leader);
- log.debug("Applied ISR change record: {}", record.toString());
+ String topicPart = topicInfo.name + "-" + record.partitionId() + " with topic ID " +
Review comment:
In the old controller, we bump up the leader epoch if the ISR is changed during the controlled shutdown. This helps prevent the shutting down broker from being added to ISR again. In the raft controller, we bump up the partitionEpoch when the ISR is changed. Do we plan to fence a fetch request with unmatched partitionEpoch to achieve the same logic? If so, do we have a jira to track that?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r618131621
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -957,22 +890,27 @@ ApiError electLeader(String topic, int partitionId, boolean unclean,
return ControllerResult.of(records, reply);
}
- int bestLeader(int[] replicas, int[] isr, boolean unclean) {
+ static boolean isGoodLeader(int[] isr, int leader) {
+ return Replicas.contains(isr, leader);
+ }
+
+ static int bestLeader(int[] replicas, int[] isr, boolean uncleanOk,
Review comment:
We use NO_LEADER in a lot of different places in the code, so I think it's reasonable to use it here. it is actually a valid value for the leader of the partition, so translating to and from OptionalInt would be awkward, I think.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617505731
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -321,22 +339,17 @@ public void replay(PartitionRecord record) {
}
PartitionControlInfo newPartInfo = new PartitionControlInfo(record);
PartitionControlInfo prevPartInfo = topicInfo.parts.get(record.partitionId());
+ String topicPart = topicInfo.name + "-" + record.partitionId();
if (prevPartInfo == null) {
- log.info("Created partition {}:{} with {}.", record.topicId(),
- record.partitionId(), newPartInfo.toString());
+ log.info("Created partition {} with {}.", topicPart, newPartInfo.toString());
Review comment:
Yes, this was an intentional change to make it clearer which topic was getting created. The ID is still available in the metadata log, of course.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617160661
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1053,61 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemoveFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemoveFromIsr);
+ int newLeader = bestLeader(partition.replicas, newIsr, false);
Review comment:
Good point. Will fix.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#issuecomment-823736138
@junrao : BTW, when do you think we should change away from a working leader to the preferred leader?
I guess when auto leader balancing is enabled and we're doing a periodic check is one such time (we haven't implemented this yet, but will soon).
Another time is when we manually trigger an election via an `electLeaders` RPC (we already implement this one).
Hmm... I wonder if there are any other scenarios.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r621639045
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
Review comment:
I think it's easier to use NO_LEADER since then I don't have to special-case the comparisons (I can just compare with NO_LEADER and it will fail when examining ISR members, for example).
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1057,83 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ /**
+ * Iterate over a sequence of partitions and generate ISR changes and/or leader
+ * changes if necessary.
+ *
+ * @param context A human-readable context string used in log4j logging.
+ * @param brokerToRemove NO_LEADER if no broker is being removed; the ID of the
+ * broker to remove from the ISR and leadership, otherwise.
+ * @param brokerToAdd NO_LEADER if no broker is being added; the ID of the
+ * broker which is now eligible to be a leader, otherwise.
+ * @param records A list of records which we will append to.
+ * @param iterator The iterator containing the partitions to examine.
+ */
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemove,
+ int brokerToAdd,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ Function<Integer, Boolean> isAcceptableLeader =
+ r -> r == brokerToAdd || clusterControl.unfenced(r);
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() +
+ " existed in isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemove);
+ int newLeader;
+ if (isGoodLeader(newIsr, partition.leader)) {
+ // If the current leader is good, don't change.
+ newLeader = partition.leader;
+ } else {
+ // Choose a new leader.
+ boolean uncleanOk = configurationControl.uncleanLeaderElectionEnabledForTopic(topic.name);
+ newLeader = bestLeader(partition.replicas, newIsr, uncleanOk, isAcceptableLeader);
Review comment:
One thing I've been considering for controlled shutdown is that perhaps we could just keep the shutting down replicas in the ISR but move the leaders. I think that would be a bit more graceful than what we have now, but we'd have to do some extra work to get there.
Anyway, let's consider this later, as you suggested...
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a change in pull request #10564: MINOR: clean up some replication code
Posted by GitBox <gi...@apache.org>.
cmccabe commented on a change in pull request #10564:
URL: https://github.com/apache/kafka/pull/10564#discussion_r617785244
##########
File path: metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
##########
@@ -1119,6 +1053,62 @@ void validateManualPartitionAssignment(List<Integer> assignment,
}
}
+ void generateLeaderAndIsrUpdates(String context,
+ int brokerToRemoveFromIsr,
+ List<ApiMessageAndVersion> records,
+ Iterator<TopicIdPartition> iterator) {
+ int oldSize = records.size();
+ while (iterator.hasNext()) {
+ TopicIdPartition topicIdPart = iterator.next();
+ TopicControlInfo topic = topics.get(topicIdPart.topicId());
+ if (topic == null) {
+ throw new RuntimeException("Topic ID " + topicIdPart.topicId() + " existed in " +
+ "isrMembers, but not in the topics map.");
+ }
+ PartitionControlInfo partition = topic.parts.get(topicIdPart.partitionId());
+ if (partition == null) {
+ throw new RuntimeException("Partition " + topicIdPart +
+ " existed in isrMembers, but not in the partitions map.");
+ }
+ int[] newIsr = Replicas.copyWithout(partition.isr, brokerToRemoveFromIsr);
+ int newLeader = Replicas.contains(newIsr, partition.leader) ? partition.leader :
+ bestLeader(partition.replicas, newIsr, false);
+ boolean unclean = newLeader != NO_LEADER && !Replicas.contains(newIsr, newLeader);
Review comment:
Yes, I think we can reuse some code here.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org