You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by cm...@apache.org on 2023/12/08 22:44:21 UTC
(kafka) branch trunk updated: KAFKA-15364: Handle log directory failure in the Controller (#14902)
This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 93b6df6173d KAFKA-15364: Handle log directory failure in the Controller (#14902)
93b6df6173d is described below
commit 93b6df6173da98a1f41f21bd90a3bada662f3e93
Author: Igor Soarez <so...@apple.com>
AuthorDate: Fri Dec 8 22:44:14 2023 +0000
KAFKA-15364: Handle log directory failure in the Controller (#14902)
When log directories fail, the broker will send a heartbeat listing the failed directories. This
PR implements processing offline directories in the controller's broker heartbeat handling. We
update broker registrations and generate leadership/ISR changes as necessary.
Co-authored-by: Colin P. McCabe <cm...@apache.org>
Reviewers: Ron Dagostino <rn...@gmail.com>
---
.../controller/ReplicationControlManager.java | 92 +++++++++++++++++++++-
.../apache/kafka/metadata/BrokerRegistration.java | 42 +++++++---
.../kafka/controller/QuorumControllerTest.java | 3 +-
.../controller/ReplicationControlManagerTest.java | 53 +++++++++++++
4 files changed, 176 insertions(+), 14 deletions(-)
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
index 56fa57ce244..1dd5e2ba162 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -114,6 +114,7 @@ import java.util.List;
import java.util.ListIterator;
import java.util.Map.Entry;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Set;
@@ -385,6 +386,11 @@ public class ReplicationControlManager {
*/
private final TimelineHashSet<TopicIdPartition> imbalancedPartitions;
+ /**
+ * A map from registered directory IDs to the partitions that are stored in that directory.
+ */
+ private final TimelineHashMap<Uuid, TimelineHashSet<TopicIdPartition>> directoriesToPartitions;
+
/**
* A ClusterDescriber which supplies cluster information to our ReplicaPlacer.
*/
@@ -420,6 +426,7 @@ public class ReplicationControlManager {
this.brokersToIsrs = new BrokersToIsrs(snapshotRegistry);
this.reassigningTopics = new TimelineHashMap<>(snapshotRegistry, 0);
this.imbalancedPartitions = new TimelineHashSet<>(snapshotRegistry, 0);
+ this.directoriesToPartitions = new TimelineHashMap<>(snapshotRegistry, 0);
}
public void replay(TopicRecord record) {
@@ -466,6 +473,7 @@ public class ReplicationControlManager {
topicInfo.parts.put(record.partitionId(), newPartInfo);
brokersToIsrs.update(record.topicId(), record.partitionId(), null,
newPartInfo.isr, NO_LEADER, newPartInfo.leader);
+ updatePartitionDirectories(record.topicId(), record.partitionId(), null, newPartInfo.directories);
updateReassigningTopicsIfNeeded(record.topicId(), record.partitionId(),
false, isReassignmentInProgress(newPartInfo));
} else if (!newPartInfo.equals(prevPartInfo)) {
@@ -475,6 +483,7 @@ public class ReplicationControlManager {
topicInfo.parts.put(record.partitionId(), newPartInfo);
brokersToIsrs.update(record.topicId(), record.partitionId(), prevPartInfo.isr,
newPartInfo.isr, prevPartInfo.leader, newPartInfo.leader);
+ updatePartitionDirectories(record.topicId(), record.partitionId(), prevPartInfo.directories, newPartInfo.directories);
updateReassigningTopicsIfNeeded(record.topicId(), record.partitionId(),
isReassignmentInProgress(prevPartInfo), isReassignmentInProgress(newPartInfo));
}
@@ -522,6 +531,7 @@ public class ReplicationControlManager {
brokersToIsrs.update(record.topicId(), record.partitionId(),
prevPartitionInfo.isr, newPartitionInfo.isr, prevPartitionInfo.leader,
newPartitionInfo.leader);
+ updatePartitionDirectories(record.topicId(), record.partitionId(), prevPartitionInfo.directories, newPartitionInfo.directories);
String topicPart = topicInfo.name + "-" + record.partitionId() + " with topic ID " +
record.topicId();
newPartitionInfo.maybeLogPartitionChange(log, topicPart, prevPartitionInfo);
@@ -569,6 +579,7 @@ public class ReplicationControlManager {
// Remove the entries for this topic in brokersToIsrs.
for (int i = 0; i < partition.isr.length; i++) {
brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]);
+ updatePartitionDirectories(topic.id, partitionId, partition.directories, null);
}
imbalancedPartitions.remove(new TopicIdPartition(record.topicId(), partitionId));
@@ -987,8 +998,8 @@ public class ReplicationControlManager {
}
// VisibleForTesting
- Set<TopicIdPartition> imbalancedPartitions() {
- return new HashSet<>(imbalancedPartitions);
+ TimelineHashSet<TopicIdPartition> imbalancedPartitions() {
+ return imbalancedPartitions;
}
boolean isElrEnabled() {
@@ -1361,6 +1372,42 @@ public class ReplicationControlManager {
brokerId, NO_LEADER, records, brokersToIsrs.partitionsWithBrokerInIsr(brokerId));
}
+ /**
+ * Generates the appropriate records to handle a list of directories being reported offline.
+ *
+ * If the reported directories include directories that were previously online, this includes
+ * a BrokerRegistrationChangeRecord and any number of PartitionChangeRecord to update
+ * leadership and ISR for partitions in those directories that were previously online.
+ *
+ * @param brokerId The broker id.
+ * @param brokerEpoch The broker epoch.
+ * @param offlineDirs The list of directories that are offline.
+ * @param records The record list to append to.
+ */
+ void handleDirectoriesOffline(
+ int brokerId,
+ long brokerEpoch,
+ List<Uuid> offlineDirs,
+ List<ApiMessageAndVersion> records
+ ) {
+ BrokerRegistration registration = clusterControl.registration(brokerId);
+ List<Uuid> newOfflineDirs = registration.directoryIntersection(offlineDirs);
+ if (!newOfflineDirs.isEmpty()) {
+ for (Uuid newOfflineDir : newOfflineDirs) {
+ TimelineHashSet<TopicIdPartition> parts = directoriesToPartitions.get(newOfflineDir);
+ Iterator<TopicIdPartition> iterator = (parts == null) ?
+ Collections.emptyIterator() : parts.iterator();
+ generateLeaderAndIsrUpdates(
+ "handleDirectoriesOffline[" + brokerId + ":" + newOfflineDir + "]",
+ brokerId, NO_LEADER, records, iterator);
+ }
+ records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
+ setBrokerId(brokerId).setBrokerEpoch(brokerEpoch).
+ setLogDirs(registration.directoryDifference(offlineDirs)),
+ (short) 2));
+ }
+ }
+
ControllerResult<ElectLeadersResponseData> electLeaders(ElectLeadersRequestData request) {
ElectionType electionType = electionType(request.electionType());
List<ApiMessageAndVersion> records = BoundedList.newArrayBacked(MAX_RECORDS_PER_USER_OP);
@@ -1497,6 +1544,9 @@ public class ReplicationControlManager {
heartbeatManager.touch(brokerId,
states.next().fenced(),
request.currentMetadataOffset());
+ if (featureControl.metadataVersion().isDirectoryAssignmentSupported()) {
+ handleDirectoriesOffline(brokerId, brokerEpoch, request.offlineLogDirs(), records);
+ }
boolean isCaughtUp = request.currentMetadataOffset() >= registerBrokerRecordOffset;
BrokerHeartbeatReply reply = new BrokerHeartbeatReply(isCaughtUp,
states.next().fenced(),
@@ -2144,6 +2194,44 @@ public class ReplicationControlManager {
return Math.min(currentMinIsr, replicationFactor);
}
+ /**
+ * Updates the directory to partition mapping for a single partition.
+ * Assignments to reserved directory IDs are ignored, since they cannot
+ * be used for directories, there's no use in maintaining a set of
+ * partitions assigned to them.
+ */
+ private void updatePartitionDirectories(
+ Uuid topicId,
+ int partitionId,
+ Uuid[] previousDirectoryIds,
+ Uuid[] newDirectoryIds
+ ) {
+ Objects.requireNonNull(topicId, "topicId cannot be null");
+ TopicIdPartition topicIdPartition = new TopicIdPartition(topicId, partitionId);
+ if (previousDirectoryIds != null) {
+ for (Uuid dir : previousDirectoryIds) {
+ if (!DirectoryId.reserved(dir)) {
+ TimelineHashSet<TopicIdPartition> partitions = directoriesToPartitions.get(dir);
+ if (partitions != null) {
+ partitions.remove(topicIdPartition);
+ if (partitions.isEmpty()) {
+ directoriesToPartitions.remove(dir);
+ }
+ }
+ }
+ }
+ }
+ if (newDirectoryIds != null) {
+ for (Uuid dir : newDirectoryIds) {
+ if (!DirectoryId.reserved(dir)) {
+ Set<TopicIdPartition> partitions = directoriesToPartitions.computeIfAbsent(dir,
+ __ -> new TimelineHashSet<>(snapshotRegistry, 0));
+ partitions.add(topicIdPartition);
+ }
+ }
+ }
+ }
+
private static final class IneligibleReplica {
private final int replicaId;
private final String reason;
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java b/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java
index e8b694c9640..6c8b0c5d608 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistration.java
@@ -123,7 +123,7 @@ public class BrokerRegistration {
}
public Builder setDirectories(List<Uuid> directories) {
- this.directories = new ArrayList<>(directories);
+ this.directories = directories;
return this;
}
@@ -159,7 +159,7 @@ public class BrokerRegistration {
private final boolean fenced;
private final boolean inControlledShutdown;
private final boolean isMigratingZkBroker;
- private final List<Uuid> sortedDirectories;
+ private final List<Uuid> directories;
private BrokerRegistration(
int id,
@@ -192,7 +192,7 @@ public class BrokerRegistration {
this.isMigratingZkBroker = isMigratingZkBroker;
directories = new ArrayList<>(directories);
directories.sort(Uuid::compareTo);
- this.sortedDirectories = Collections.unmodifiableList(directories);
+ this.directories = Collections.unmodifiableList(directories);
}
public static BrokerRegistration fromRecord(RegisterBrokerRecord record) {
@@ -265,11 +265,31 @@ public class BrokerRegistration {
}
public List<Uuid> directories() {
- return sortedDirectories;
+ return directories;
}
public boolean hasOnlineDir(Uuid dir) {
- return DirectoryId.isOnline(dir, sortedDirectories);
+ return DirectoryId.isOnline(dir, directories);
+ }
+
+ public List<Uuid> directoryIntersection(List<Uuid> otherDirectories) {
+ List<Uuid> results = new ArrayList<>();
+ for (Uuid directory : directories) {
+ if (otherDirectories.contains(directory)) {
+ results.add(directory);
+ }
+ }
+ return results;
+ }
+
+ public List<Uuid> directoryDifference(List<Uuid> otherDirectories) {
+ List<Uuid> results = new ArrayList<>();
+ for (Uuid directory : directories) {
+ if (!otherDirectories.contains(directory)) {
+ results.add(directory);
+ }
+ }
+ return results;
}
public ApiMessageAndVersion toRecord(ImageWriterOptions options) {
@@ -296,8 +316,8 @@ public class BrokerRegistration {
}
}
- if (sortedDirectories.isEmpty() || options.metadataVersion().isDirectoryAssignmentSupported()) {
- registrationRecord.setLogDirs(sortedDirectories);
+ if (directories.isEmpty() || options.metadataVersion().isDirectoryAssignmentSupported()) {
+ registrationRecord.setLogDirs(directories);
} else {
options.handleLoss("the online log directories of one or more brokers");
}
@@ -325,7 +345,7 @@ public class BrokerRegistration {
@Override
public int hashCode() {
return Objects.hash(id, epoch, incarnationId, listeners, supportedFeatures,
- rack, fenced, inControlledShutdown, isMigratingZkBroker, sortedDirectories);
+ rack, fenced, inControlledShutdown, isMigratingZkBroker, directories);
}
@Override
@@ -341,7 +361,7 @@ public class BrokerRegistration {
other.fenced == fenced &&
other.inControlledShutdown == inControlledShutdown &&
other.isMigratingZkBroker == isMigratingZkBroker &&
- other.sortedDirectories.equals(sortedDirectories);
+ other.directories.equals(directories);
}
@Override
@@ -363,7 +383,7 @@ public class BrokerRegistration {
bld.append(", fenced=").append(fenced);
bld.append(", inControlledShutdown=").append(inControlledShutdown);
bld.append(", isMigratingZkBroker=").append(isMigratingZkBroker);
- bld.append(", directories=").append(sortedDirectories);
+ bld.append(", directories=").append(directories);
bld.append(")");
return bld.toString();
}
@@ -388,7 +408,7 @@ public class BrokerRegistration {
newFenced,
newInControlledShutdownChange,
isMigratingZkBroker,
- sortedDirectories
+ directories
);
}
}
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
index 423963cb3ae..cac74c6a2d3 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -24,6 +24,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -464,7 +465,7 @@ public class QuorumControllerTest {
sendBrokerHeartbeatToUnfenceBrokers(active, allBrokers, brokerEpochs);
// Let the unfenced broker, 3, join the ISR partition 2
- Set<TopicIdPartition> imbalancedPartitions = active.replicationControl().imbalancedPartitions();
+ Set<TopicIdPartition> imbalancedPartitions = new HashSet<>(active.replicationControl().imbalancedPartitions());
assertEquals(1, imbalancedPartitions.size());
int imbalancedPartitionId = imbalancedPartitions.iterator().next().partitionId();
PartitionRegistration partitionRegistration = active.replicationControl().getPartition(topicIdFoo, imbalancedPartitionId);
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
index 9d3b3e891fe..c41e629fbc7 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -68,6 +68,7 @@ import org.apache.kafka.common.metadata.RemoveTopicRecord;
import org.apache.kafka.common.metadata.TopicRecord;
import org.apache.kafka.common.metadata.PartitionRecord;
import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ApiMessage;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AlterPartitionRequest;
import org.apache.kafka.common.requests.ApiError;
@@ -2932,6 +2933,54 @@ public class ReplicationControlManagerTest {
RecordTestUtils.iteratorToSet(ctx.replicationControl.brokersToIsrs().iterator(2, true)));
}
+ @Test
+ void testHandleDirectoriesOffline() throws Exception {
+ ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();
+ int b1 = 101, b2 = 102;
+ Uuid dir1b1 = Uuid.fromString("suitdzfTTdqoWcy8VqmkUg");
+ Uuid dir2b1 = Uuid.fromString("yh3acnzGSeurSTj8aIhOjw");
+ Uuid dir1b2 = Uuid.fromString("OmpmJ8RjQliQlEFht56DwQ");
+ Uuid dir2b2 = Uuid.fromString("w05baLpsT5Oz0LvKTKXoDw");
+ ctx.registerBrokersWithDirs(b1, asList(dir1b1, dir2b1), b2, asList(dir1b2, dir2b2));
+ ctx.unfenceBrokers(b1, b2);
+ Uuid topicA = ctx.createTestTopic("a", new int[][]{new int[]{b1, b2}, new int[]{b1, b2}}).topicId();
+ Uuid topicB = ctx.createTestTopic("b", new int[][]{new int[]{b1, b2}, new int[]{b1, b2}}).topicId();
+ ctx.assignReplicasToDirs(b1, new HashMap<TopicIdPartition, Uuid>() {{
+ put(new TopicIdPartition(topicA, 0), dir1b1);
+ put(new TopicIdPartition(topicA, 1), dir2b1);
+ put(new TopicIdPartition(topicB, 0), dir1b1);
+ put(new TopicIdPartition(topicB, 1), dir2b1);
+ }});
+ ctx.assignReplicasToDirs(b2, new HashMap<TopicIdPartition, Uuid>() {{
+ put(new TopicIdPartition(topicA, 0), dir1b2);
+ put(new TopicIdPartition(topicA, 1), dir2b2);
+ put(new TopicIdPartition(topicB, 0), dir1b2);
+ put(new TopicIdPartition(topicB, 1), dir2b2);
+ }});
+ List<ApiMessageAndVersion> records = new ArrayList<>();
+ ctx.replicationControl.handleDirectoriesOffline(b1, defaultBrokerEpoch(b1), asList(
+ dir1b1,
+ dir1b2 // should not cause update to dir1b2 as it's not registered to b1
+ ), records);
+ assertEquals(
+ singletonList(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord()
+ .setBrokerId(b1).setBrokerEpoch(defaultBrokerEpoch(b1))
+ .setLogDirs(singletonList(dir2b1)), (short) 2)),
+ filter(records, BrokerRegistrationChangeRecord.class)
+ );
+ short partitionChangeRecordVersion = ctx.featureControl.metadataVersion().partitionChangeRecordVersion();
+ assertEquals(
+ sortPartitionChangeRecords(asList(
+ new ApiMessageAndVersion(new PartitionChangeRecord().setTopicId(topicA).setPartitionId(0)
+ .setLeader(b2).setIsr(singletonList(b2)), partitionChangeRecordVersion),
+ new ApiMessageAndVersion(new PartitionChangeRecord().setTopicId(topicB).setPartitionId(0)
+ .setLeader(b2).setIsr(singletonList(b2)), partitionChangeRecordVersion)
+ )),
+ sortPartitionChangeRecords(filter(records, PartitionChangeRecord.class))
+ );
+ assertEquals(3, records.size());
+ }
+
/**
* Sorts {@link PartitionChangeRecord} by topic ID and partition ID,
* so that the order of the records is deterministic, and can be compared.
@@ -2944,4 +2993,8 @@ public class ReplicationControlManagerTest {
}));
return records;
}
+
+ private static List<ApiMessageAndVersion> filter(List<ApiMessageAndVersion> records, Class<? extends ApiMessage> clazz) {
+ return records.stream().filter(r -> clazz.equals(r.message().getClass())).collect(Collectors.toList());
+ }
}