You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by js...@apache.org on 2022/10/18 22:33:37 UTC
[kafka] branch 3.3 updated: KAFKA-14300; Generate snapshot after repeated controller resign (#12747)
This is an automated email from the ASF dual-hosted git repository.
jsancio pushed a commit to branch 3.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.3 by this push:
new b789cb25799 KAFKA-14300; Generate snapshot after repeated controller resign (#12747)
b789cb25799 is described below
commit b789cb25799f815debf972854c00c84e85873e91
Author: José Armando García Sancio <js...@users.noreply.github.com>
AuthorDate: Tue Oct 18 15:09:20 2022 -0700
KAFKA-14300; Generate snapshot after repeated controller resign (#12747)
Setting the `committedBytesSinceLastSnapshot` to 0 when resigning can cause the controller to not generate a snapshot after `snapshotMaxNewRecordBytes` committed bytes have been replayed.
This change fixes that by simply not resetting the counter during resignation. This is correct because the counter tracks the number of committed bytes replayed but not included in the latest snapshot. In other words, reverting the last committed state does not invalidate this value.
Reviewers: Colin Patrick McCabe <cm...@apache.org>
---
metadata/.jqwik-database | Bin 0 -> 4 bytes
.../apache/kafka/controller/QuorumController.java | 43 ++++++++-----
.../apache/kafka/timeline/SnapshotRegistry.java | 2 +-
.../kafka/controller/QuorumControllerTest.java | 67 +++++++++++++++++++++
.../org/apache/kafka/metalog/LocalLogManager.java | 3 +
5 files changed, 98 insertions(+), 17 deletions(-)
diff --git a/metadata/.jqwik-database b/metadata/.jqwik-database
new file mode 100644
index 00000000000..711006c3d3b
Binary files /dev/null and b/metadata/.jqwik-database differ
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 2255c741535..ef758c71598 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -979,11 +979,15 @@ public final class QuorumController implements Controller {
i++;
}
}
- updateLastCommittedState(offset, epoch, batch.appendTimestamp());
- processedRecordsSize += batch.sizeInBytes();
+ updateLastCommittedState(
+ offset,
+ epoch,
+ batch.appendTimestamp(),
+ committedBytesSinceLastSnapshot + batch.sizeInBytes()
+ );
}
- maybeGenerateSnapshot(processedRecordsSize);
+ maybeGenerateSnapshot();
} finally {
reader.close();
}
@@ -1038,10 +1042,10 @@ public final class QuorumController implements Controller {
updateLastCommittedState(
reader.lastContainedLogOffset(),
reader.lastContainedLogEpoch(),
- reader.lastContainedLogTimestamp()
+ reader.lastContainedLogTimestamp(),
+ 0
);
snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
- newBytesSinceLastSnapshot = 0L;
authorizer.ifPresent(a -> a.loadSnapshot(aclControlManager.idToAcl()));
} finally {
reader.close();
@@ -1193,10 +1197,16 @@ public final class QuorumController implements Controller {
}
}
- private void updateLastCommittedState(long offset, int epoch, long timestamp) {
+ private void updateLastCommittedState(
+ long offset,
+ int epoch,
+ long timestamp,
+ long bytesSinceLastSnapshot
+ ) {
lastCommittedOffset = offset;
lastCommittedEpoch = epoch;
lastCommittedTimestamp = timestamp;
+ committedBytesSinceLastSnapshot = bytesSinceLastSnapshot;
controllerMetrics.setLastCommittedRecordOffset(offset);
if (!isActiveController()) {
@@ -1222,7 +1232,6 @@ public final class QuorumController implements Controller {
}
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
authorizer.ifPresent(a -> a.loadSnapshot(aclControlManager.idToAcl()));
- newBytesSinceLastSnapshot = 0L;
updateWriteOffset(-1);
clusterControl.deactivate();
cancelMaybeFenceReplicas();
@@ -1444,9 +1453,8 @@ public final class QuorumController implements Controller {
}
}
- private void maybeGenerateSnapshot(long batchSizeInBytes) {
- newBytesSinceLastSnapshot += batchSizeInBytes;
- if (newBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
+ private void maybeGenerateSnapshot() {
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
snapshotGeneratorManager.generator == null
) {
if (!isActiveController()) {
@@ -1457,10 +1465,14 @@ public final class QuorumController implements Controller {
}
log.info("Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot.",
- lastCommittedEpoch, lastCommittedOffset, newBytesSinceLastSnapshot);
+ lastCommittedEpoch, lastCommittedOffset, committedBytesSinceLastSnapshot);
- snapshotGeneratorManager.createSnapshotGenerator(lastCommittedOffset, lastCommittedEpoch, lastCommittedTimestamp);
- newBytesSinceLastSnapshot = 0;
+ snapshotGeneratorManager.createSnapshotGenerator(
+ lastCommittedOffset,
+ lastCommittedEpoch,
+ lastCommittedTimestamp
+ );
+ committedBytesSinceLastSnapshot = 0;
}
}
@@ -1471,8 +1483,7 @@ public final class QuorumController implements Controller {
snapshotGeneratorManager.cancel();
snapshotRegistry.reset();
- newBytesSinceLastSnapshot = 0;
- updateLastCommittedState(-1, -1, -1);
+ updateLastCommittedState(-1, -1, -1, 0);
}
/**
@@ -1645,7 +1656,7 @@ public final class QuorumController implements Controller {
/**
* Number of bytes processed through handling commits since the last snapshot was generated.
*/
- private long newBytesSinceLastSnapshot = 0;
+ private long committedBytesSinceLastSnapshot = 0;
/**
* How long to delay partition leader balancing operations.
diff --git a/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java b/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java
index 49d62fd2117..0df9c8b8f56 100644
--- a/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java
+++ b/metadata/src/main/java/org/apache/kafka/timeline/SnapshotRegistry.java
@@ -237,7 +237,7 @@ public class SnapshotRegistry {
} else {
snapshot.erase();
}
- log.debug("Deleting snapshot {}", snapshot.epoch());
+ log.debug("Deleting in-memory snapshot {}", snapshot.epoch());
snapshots.remove(snapshot.epoch(), snapshot);
}
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 a5e7258a724..90807a2e2a7 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -821,6 +821,73 @@ public class QuorumControllerTest {
}
}
+ @Test
+ public void testSnapshotAfterRepeatedResign() throws Throwable {
+ final int numBrokers = 4;
+ final int maxNewRecordBytes = 1000;
+ Map<Integer, Long> brokerEpochs = new HashMap<>();
+ try (
+ LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).
+ build();
+ QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).
+ setControllerBuilderInitializer(controllerBuilder -> {
+ controllerBuilder.setConfigSchema(SCHEMA);
+ controllerBuilder.setSnapshotMaxNewRecordBytes(maxNewRecordBytes);
+ }).
+ build();
+ ) {
+ QuorumController active = controlEnv.activeController();
+ for (int i = 0; i < numBrokers; i++) {
+ BrokerRegistrationReply reply = active.registerBroker(ANONYMOUS_CONTEXT,
+ new BrokerRegistrationRequestData().
+ setBrokerId(i).
+ setRack(null).
+ setClusterId(active.clusterId()).
+ setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_3_IV3)).
+ setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwB" + i)).
+ setListeners(new ListenerCollection(Arrays.asList(new Listener().
+ setName("PLAINTEXT").setHost("localhost").
+ setPort(9092 + i)).iterator()))).get();
+ brokerEpochs.put(i, reply.epoch());
+ assertEquals(new BrokerHeartbeatReply(true, false, false, false),
+ active.processBrokerHeartbeat(ANONYMOUS_CONTEXT, new BrokerHeartbeatRequestData().
+ setWantFence(false).setBrokerEpoch(brokerEpochs.get(i)).
+ setBrokerId(i).setCurrentMetadataOffset(100000L)).get());
+ }
+
+ assertTrue(logEnv.appendedBytes() < maxNewRecordBytes,
+ String.format("%s appended bytes is not less than %s max new record bytes",
+ logEnv.appendedBytes(),
+ maxNewRecordBytes));
+
+ // Keep creating topic and resign leader until we reached the max bytes limit
+ int counter = 0;
+ while (logEnv.appendedBytes() < maxNewRecordBytes) {
+ active = controlEnv.activeController();
+
+ counter += 1;
+ String topicName = String.format("foo-%s", counter);
+ active.createTopics(ANONYMOUS_CONTEXT, new CreateTopicsRequestData().setTopics(
+ new CreatableTopicCollection(Collections.singleton(
+ new CreatableTopic().setName(topicName).setNumPartitions(-1).
+ setReplicationFactor((short) -1).
+ setAssignments(new CreatableReplicaAssignmentCollection(
+ Arrays.asList(new CreatableReplicaAssignment().
+ setPartitionIndex(0).
+ setBrokerIds(Arrays.asList(0, 1, 2)),
+ new CreatableReplicaAssignment().
+ setPartitionIndex(1).
+ setBrokerIds(Arrays.asList(1, 2, 0))).
+ iterator()))).iterator())),
+ Collections.singleton(topicName)).get(60, TimeUnit.SECONDS);
+
+ LocalLogManager activeLocalLogManager = logEnv.logManagers().get(active.nodeId());
+ activeLocalLogManager.resign(activeLocalLogManager.leaderAndEpoch().epoch());
+ }
+ logEnv.waitForLatestSnapshot();
+ }
+ }
+
private SnapshotReader<ApiMessageAndVersion> createSnapshotReader(RawSnapshotReader reader) {
return RecordsSnapshotReader.of(
reader,
diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
index e24d86bd873..6457ede9ba2 100644
--- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
+++ b/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java
@@ -406,7 +406,10 @@ public final class LocalLogManager implements RaftClient<ApiMessageAndVersion>,
}
void handleLeaderChange(long offset, LeaderAndEpoch leader) {
+ // Simulate KRaft implementation by first bumping the epoch before assigning a leader
+ listener.handleLeaderChange(new LeaderAndEpoch(OptionalInt.empty(), leader.epoch()));
listener.handleLeaderChange(leader);
+
notifiedLeader = leader;
this.offset = offset;
}