Skip to content

KAFKA-20295: Use current quorum voters for feature validation #23038

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lh0156 wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

KAFKA-20295: Use current quorum voters for feature validation #23038

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions metadata/src/main/java/org/apache/kafka/controller/ClusterFeatureSupportDescriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,23 @@
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;


public interface ClusterFeatureSupportDescriber {
Iterator<Entry<Integer, Map<String, VersionRange>>> brokerSupported();
Iterator<Entry<Integer, Map<String, VersionRange>>> controllerSupported();

/**
* The IDs of the controllers which are currently members of the quorum.
*
* <p>This is separate from {@link #controllerSupported()} because a controller
* registration can outlive the controller's membership in a dynamic quorum.
*
* @return the current quorum controller IDs, or an empty set when the caller
* should use its static quorum configuration
*/
default Set<Integer> controllerIds() {
return Set.of();
}
}
10 changes: 9 additions & 1 deletion metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.function.Consumer;

import static org.apache.kafka.common.metadata.MetadataRecordType.FEATURE_LEVEL_RECORD;
Expand Down Expand Up @@ -341,10 +342,17 @@ private Optional<String> reasonNotSupported(
HashSet<Integer> foundControllers = new HashSet<>();
foundControllers.add(quorumFeatures.nodeId());
if (metadataVersionOrThrow().isControllerRegistrationSupported()) {
Set<Integer> controllerIds = clusterSupportDescriber.controllerIds();
if (controllerIds.isEmpty()) {
controllerIds = Set.copyOf(quorumFeatures.quorumNodeIds());
}
for (Iterator<Entry<Integer, Map<String, VersionRange>>> iter =
clusterSupportDescriber.controllerSupported();
iter.hasNext(); ) {
Entry<Integer, Map<String, VersionRange>> entry = iter.next();
if (!controllerIds.contains(entry.getKey())) {
continue;
}
if (entry.getKey() == quorumFeatures.nodeId()) {
// No need to re-check the features supported by this controller, since we
// already checked that above.
Expand All @@ -357,7 +365,7 @@ private Optional<String> reasonNotSupported(
foundControllers.add(entry.getKey());
numControllersChecked++;
}
for (int id : quorumFeatures.quorumNodeIds()) {
for (int id : controllerIds) {
if (!foundControllers.contains(id)) {
return Optional.of("controller " + id + " has not registered, and may not " +
"support this feature");
Expand Down
5 changes: 5 additions & 0 deletions metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -527,6 +527,11 @@ public Iterator<Entry<Integer, Map<String, VersionRange>>> brokerSupported() {
public Iterator<Entry<Integer, Map<String, VersionRange>>> controllerSupported() {
return clusterControl.controllerSupportedFeatures();
}

@Override
public Set<Integer> controllerIds() {
return raftClient.voterIds();
}
}

class PeriodicTaskControlManagerQueueAccessor implements PeriodicTaskControlManager.QueueAccessor {
Expand Down
75 changes: 75 additions & 0 deletions metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,6 +176,14 @@ public void testReplayKraftVersionFeatureLevel() {
static ClusterFeatureSupportDescriber createFakeClusterFeatureSupportDescriber(
List<Map.Entry<Integer, Map<String, VersionRange>>> brokerRanges,
List<Map.Entry<Integer, Map<String, VersionRange>>> controllerRanges
) {
return createFakeClusterFeatureSupportDescriber(brokerRanges, controllerRanges, Set.of());
}

static ClusterFeatureSupportDescriber createFakeClusterFeatureSupportDescriber(
List<Map.Entry<Integer, Map<String, VersionRange>>> brokerRanges,
List<Map.Entry<Integer, Map<String, VersionRange>>> controllerRanges,
Set<Integer> controllerIds
) {
return new ClusterFeatureSupportDescriber() {
@Override
Expand All @@ -187,9 +195,76 @@ public Iterator<Map.Entry<Integer, Map<String, VersionRange>>> brokerSupported()
public Iterator<Map.Entry<Integer, Map<String, VersionRange>>> controllerSupported() {
return controllerRanges.iterator();
}

@Override
public Set<Integer> controllerIds() {
return controllerIds;
}
};
}

@Test
public void testFeatureUpgradeIgnoresRemovedControllerRegistration() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(new QuorumFeatures(
0,
QuorumFeatures.defaultSupportedFeatureMap(true),
List.of(0, 1, 2))).
setClusterFeatureSupportDescriber(createFakeClusterFeatureSupportDescriber(
List.of(),
List.of(
new SimpleImmutableEntry<>(1, Map.of(
MetadataVersion.FEATURE_NAME,
VersionRange.of(MetadataVersion.IBP_3_7_IV0.featureLevel(),
MetadataVersion.IBP_4_3_IV0.featureLevel()))),
new SimpleImmutableEntry<>(2, Map.of(
MetadataVersion.FEATURE_NAME,
VersionRange.of(MetadataVersion.IBP_3_7_IV0.featureLevel(),
MetadataVersion.IBP_4_2_IV1.featureLevel())))
),
Set.of(0, 1))).
build();
manager.replay(new FeatureLevelRecord().setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(MetadataVersion.IBP_3_7_IV0.featureLevel()));

ControllerResult<ApiError> result = manager.updateFeatures(
Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_4_3_IV0.featureLevel()),
Map.of(),
false,
0);

assertEquals(ApiError.NONE, result.response());
}

@Test
public void testFeatureUpgradeRequiresCurrentControllerRegistration() {
FeatureControlManager manager = new FeatureControlManager.Builder().
setQuorumFeatures(new QuorumFeatures(
0,
QuorumFeatures.defaultSupportedFeatureMap(true),
List.of(0, 1))).
setClusterFeatureSupportDescriber(createFakeClusterFeatureSupportDescriber(
List.of(),
List.of(new SimpleImmutableEntry<>(1, Map.of(
MetadataVersion.FEATURE_NAME,
VersionRange.of(MetadataVersion.IBP_3_7_IV0.featureLevel(),
MetadataVersion.IBP_4_3_IV0.featureLevel())))),
Set.of(0, 1, 2))).
build();
manager.replay(new FeatureLevelRecord().setName(MetadataVersion.FEATURE_NAME).
setFeatureLevel(MetadataVersion.IBP_3_7_IV0.featureLevel()));

ControllerResult<ApiError> result = manager.updateFeatures(
Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_4_3_IV0.featureLevel()),
Map.of(),
false,
0);

assertEquals(new ApiError(Errors.INVALID_UPDATE_VERSION,
"Invalid update version 30 for feature metadata.version. controller 2 has not " +
"registered, and may not support this feature"), result.response());
}

@Test
public void testUpdateFeaturesErrorCases() {
LogContext logContext = new LogContext();
Expand Down
6 changes: 6 additions & 0 deletions metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
Expand Down Expand Up @@ -765,6 +766,11 @@ public Optional<Node> voterNode(int id, ListenerName listenerName) {
return Optional.empty();
}

@Override
public Set<Integer> voterIds() {
return Set.copyOf(shared.raftClients.keySet());
}

public List<RaftClient.Listener<ApiMessageAndVersion>> listeners() {
final CompletableFuture<List<RaftClient.Listener<ApiMessageAndVersion>>> future = new CompletableFuture<>();
eventQueue.append(() ->
Expand Down
6 changes: 6 additions & 0 deletions metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;

Expand Down Expand Up @@ -80,6 +81,11 @@ public Optional<Node> voterNode(int id, ListenerName listenerName) {
return Optional.empty();
}

@Override
public Set<Integer> voterIds() {
return Set.of();
}

@Override
public long prepareAppend(int epoch, List<ApiMessageAndVersion> records) {
return 0;
Expand Down
5 changes: 5 additions & 0 deletions raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3900,6 +3900,11 @@ public Optional<Node> voterNode(int id, ListenerName listenerName) {
return partitionState.lastVoterSet().voterNode(id, listenerName);
}

@Override
public Set<Integer> voterIds() {
return partitionState.lastVoterSet().voterIds();
}

// Visible only for test
QuorumState quorum() {
// It's okay to return null since this method is only called by tests
Expand Down
8 changes: 8 additions & 0 deletions raft/src/main/java/org/apache/kafka/raft/RaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.CompletionStage;

public interface RaftClient<T> extends AutoCloseable {
Expand Down Expand Up @@ -152,6 +153,13 @@ default void beginShutdown() {}
*/
Optional<Node> voterNode(int id, ListenerName listenerName);

/**
* Return the IDs of the voters in the current quorum.
*
* @return the current voter IDs
*/
Set<Integer> voterIds();

/**
* Prepare a list of records to be appended to the log.
*
Expand Down
2 changes: 2 additions & 0 deletions raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1124,6 +1124,7 @@ public void testRemoveVoter() throws Exception {
int epoch = context.currentEpoch();

assertTrue(context.client.quorum().isVoter(follower2));
assertEquals(voters.voterIds(), context.client.voterIds());

checkLeaderMetricValues(3, 0, 0, context);

Expand All @@ -1144,6 +1145,7 @@ public void testRemoveVoter() throws Exception {

// follower2 should not be a voter in the latest voter set
assertFalse(context.client.quorum().isVoter(follower2));
assertEquals(Set.of(local.id(), follower1.id()), context.client.voterIds());
checkLeaderMetricValues(2, 1, 1, context);

// Send a FETCH to increase the HWM and commit the new voter set
Expand Down