Skip to content

MINOR: only allow certain operations when in KRaft premigration #13003

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

Closed
cmccabe wants to merge 1 commit into from
Closed

MINOR: only allow certain operations when in KRaft premigration #13003

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
117 changes: 83 additions & 34 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 @@ -106,6 +106,7 @@
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
Expand All @@ -127,6 +128,8 @@

import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.kafka.controller.QuorumController.ControllerOperationFlag.DOES_NOT_UPDATE_QUEUE_TIME;
import static org.apache.kafka.controller.QuorumController.ControllerOperationFlag.RUNS_IN_PREMIGRATION;


/**
Expand Down Expand Up @@ -724,6 +727,26 @@ <T> CompletableFuture<T> appendReadEvent(
return event.future();
}

enum ControllerOperationFlag {
/**
* A flag that signifies that this operation should not update the event queue time metric.
* We use this when the event was not appended to the queue.
*/
DOES_NOT_UPDATE_QUEUE_TIME,

/**
* A flag that signifies that this operation can be processed when in pre-migration mode.
* Operations without this flag will always return NOT_CONTROLLER when invoked in premigration
* mode.
* <p>
* In pre-migration mode, we are still waiting to load the metadata from Apache
* ZooKeeper into the metadata log. Therefore, the metadata log is mostly empty,
* even though the cluster really does have metadata. Very few operations should
* use this flag.
*/
RUNS_IN_PREMIGRATION
}

interface ControllerWriteOperation<T> {
/**
* Generate the metadata records needed to implement this controller write
Expand Down Expand Up @@ -761,19 +784,19 @@ class ControllerWriteEvent<T> implements EventQueue.Event, DeferredEvent {
private final CompletableFuture<T> future;
private final ControllerWriteOperation<T> op;
private final long eventCreatedTimeNs = time.nanoseconds();
private final boolean deferred;
private final EnumSet<ControllerOperationFlag> flags;
private OptionalLong startProcessingTimeNs = OptionalLong.empty();
private ControllerResultAndOffset<T> resultAndOffset;

ControllerWriteEvent(String name, ControllerWriteOperation<T> op) {
this(name, op, false);
}

ControllerWriteEvent(String name, ControllerWriteOperation<T> op, boolean deferred) {
ControllerWriteEvent(
String name,
ControllerWriteOperation<T> op,
EnumSet<ControllerOperationFlag> flags
) {
this.name = name;
this.future = new CompletableFuture<T>();
this.op = op;
this.deferred = deferred;
this.flags = flags;
this.resultAndOffset = null;
}

Expand All @@ -784,13 +807,16 @@ CompletableFuture<T> future() {
@Override
public void run() throws Exception {
long now = time.nanoseconds();
if (!deferred) {
// We exclude deferred events from the event queue time metric to prevent
// incorrectly including the deferral time in the queue time.
if (!flags.contains(DOES_NOT_UPDATE_QUEUE_TIME)) {
controllerMetrics.updateEventQueueTime(NANOSECONDS.toMillis(now - eventCreatedTimeNs));
}
int controllerEpoch = curClaimEpoch;
if (!isActiveController()) {
if (!isActiveController(controllerEpoch)) {
throw newNotControllerException();
}
if (inPremigrationMode && !flags.contains(RUNS_IN_PREMIGRATION)) {
log.info("Cannot run write operation {} in premigration mode. Returning " +
"NOT_CONTROLLER.", name);
throw newNotControllerException();
}
startProcessingTimeNs = OptionalLong.of(now);
Expand Down Expand Up @@ -951,10 +977,21 @@ static long appendRecords(
}
}

<T> CompletableFuture<T> appendWriteEvent(String name,
OptionalLong deadlineNs,
ControllerWriteOperation<T> op) {
ControllerWriteEvent<T> event = new ControllerWriteEvent<>(name, op);
<T> CompletableFuture<T> appendWriteEvent(
String name,
OptionalLong deadlineNs,
ControllerWriteOperation<T> op
) {
return appendWriteEvent(name, deadlineNs, op, EnumSet.noneOf(ControllerOperationFlag.class));
}

<T> CompletableFuture<T> appendWriteEvent(
String name,
OptionalLong deadlineNs,
ControllerWriteOperation<T> op,
EnumSet<ControllerOperationFlag> flags
) {
ControllerWriteEvent<T> event = new ControllerWriteEvent<>(name, op, flags);
if (deadlineNs.isPresent()) {
queue.appendWithDeadline(deadlineNs.getAsLong(), event);
} else {
Expand All @@ -969,7 +1006,6 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
appendRaftEvent("handleCommit[baseOffset=" + reader.baseOffset() + "]", () -> {
try {
maybeCompleteAuthorizerInitialLoad();
long processedRecordsSize = 0;
boolean isActive = isActiveController();
while (reader.hasNext()) {
Batch<ApiMessageAndVersion> batch = reader.next();
Expand Down Expand Up @@ -1169,7 +1205,11 @@ private void maybeCompleteAuthorizerInitialLoad() {
}

private boolean isActiveController() {
return curClaimEpoch != -1;
return isActiveController(curClaimEpoch);
}

private boolean isActiveController(int claimEpoch) {
return claimEpoch != -1;
}

private void updateWriteOffset(long offset) {
Expand Down Expand Up @@ -1208,7 +1248,8 @@ private void claim(int epoch) {
// of the queue rather than the end (hence prepend rather than append). It's also
// important not to use prepend for anything else, to preserve the ordering here.
queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
new CompleteActivationEvent()));
new CompleteActivationEvent(),
EnumSet.of(DOES_NOT_UPDATE_QUEUE_TIME, RUNS_IN_PREMIGRATION)));
} catch (Throwable e) {
fatalFaultHandler.handleFault("exception while claiming leadership", e);
}
Expand Down Expand Up @@ -1289,8 +1330,15 @@ private void renounce() {
}
}

private <T> void scheduleDeferredWriteEvent(String name, long deadlineNs,
ControllerWriteOperation<T> op) {
private <T> void scheduleDeferredWriteEvent(
String name,
long deadlineNs,
ControllerWriteOperation<T> op,
EnumSet<ControllerOperationFlag> flags
) {
if (!flags.contains(DOES_NOT_UPDATE_QUEUE_TIME)) {
throw new RuntimeException("deferred events should not update the queue time.");
}
ControllerWriteEvent<T> event = new ControllerWriteEvent<>(name, op, true);
queue.scheduleDeferred(name, new EarliestDeadlineFunction(deadlineNs), event);
event.future.exceptionally(e -> {
Expand All @@ -1307,7 +1355,7 @@ private <T> void scheduleDeferredWriteEvent(String name, long deadlineNs,
log.error("Unexpected exception while executing deferred write event {}. " +
"Rescheduling for a minute from now.", name, e);
scheduleDeferredWriteEvent(name,
deadlineNs + NANOSECONDS.convert(1, TimeUnit.MINUTES), op);
deadlineNs + NANOSECONDS.convert(1, TimeUnit.MINUTES), op, flags);
return null;
});
}
Expand All @@ -1320,13 +1368,15 @@ private void rescheduleMaybeFenceStaleBrokers() {
cancelMaybeFenceReplicas();
return;
}
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs, () -> {
ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
// This following call ensures that if there are multiple brokers that
// are currently stale, then fencing for them is scheduled immediately
rescheduleMaybeFenceStaleBrokers();
return result;
});
scheduleDeferredWriteEvent(MAYBE_FENCE_REPLICAS, nextCheckTimeNs,
() -> {
ControllerResult<Void> result = replicationControl.maybeFenceOneStaleBroker();
// This following call ensures that if there are multiple brokers that
// are currently stale, then fencing for them is scheduled immediately
rescheduleMaybeFenceStaleBrokers();
return result;
},
EnumSet.of(DOES_NOT_UPDATE_QUEUE_TIME, RUNS_IN_PREMIGRATION));
}

private void cancelMaybeFenceReplicas() {
Expand All @@ -1337,9 +1387,9 @@ private void cancelMaybeFenceReplicas() {

private void maybeScheduleNextBalancePartitionLeaders() {
if (imbalancedScheduled != ImbalanceSchedule.SCHEDULED &&
leaderImbalanceCheckIntervalNs.isPresent() &&
replicationControl.arePartitionLeadersImbalanced()) {

leaderImbalanceCheckIntervalNs.isPresent() &&
replicationControl.arePartitionLeadersImbalanced() &&
inPremigrationMode == false) {
log.debug(
"Scheduling write event for {} because scheduled ({}), checkIntervalNs ({}) and isImbalanced ({})",
MAYBE_BALANCE_PARTITION_LEADERS,
Expand All @@ -1364,7 +1414,7 @@ private void maybeScheduleNextBalancePartitionLeaders() {
// generated by a ControllerWriteEvent have been applied.

return result;
}, true);
}, EnumSet.of(DOES_NOT_UPDATE_QUEUE_TIME));

long delayNs = time.nanoseconds();
if (imbalancedScheduled == ImbalanceSchedule.DEFERRED) {
Expand Down Expand Up @@ -1412,8 +1462,7 @@ private void maybeScheduleNextWriteNoOpRecord() {
null
);
},
true
);
EnumSet.of(DOES_NOT_UPDATE_QUEUE_TIME, RUNS_IN_PREMIGRATION));

long delayNs = time.nanoseconds() + maxIdleIntervalNs.getAsLong();
queue.scheduleDeferred(WRITE_NO_OP_RECORD, new EarliestDeadlineFunction(delayNs), event);
Expand Down