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 2022/10/17 21:32:29 UTC
[GitHub] [kafka] jsancio opened a new pull request, #12761: KAFKA-14307; Controller time-based snapshots
jsancio opened a new pull request, #12761:
URL: https://github.com/apache/kafka/pull/12761
TODO!!
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026965577
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
It is the oldest timestamp in the log that is not included in a snapshot. I was thinking that `oldestCommittedLogAppendTimestamp` could mislead the reader if they don't read the description of that variable:
```java
/**
* Timestamp for the oldest record that was committed but not included in a snapshot.
*/
private long oldestCommittedLogOnlyAppendTimestamp = Long.MAX_VALUE;
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r998837515
##########
raft/src/main/java/org/apache/kafka/raft/OffsetAndEpoch.java:
##########
@@ -25,6 +25,14 @@ public OffsetAndEpoch(long offset, int epoch) {
this.epoch = epoch;
}
+ public long offset() {
+ return offset;
+ }
+
+ public int epoch() {
Review Comment:
The fields are public already. I know that the methods are sometimes nicer for lambda expressions. Should we go ahead and make the fields private?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028360492
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,15 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the time elapsed, see " +
+ s"the <code>$MetadataSnapshotMaxIntervalMsProp</code> configuration."
+ val MetadataSnapshotMaxIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
+ "if there are committed records in the log that are not included in the latest snapshot. A value of zero disables " +
+ s"time based snapshot generation. The default value is ${Defaults.MetadataSnapshotMaxIntervalMs}. To geneate " +
Review Comment:
typo: "geneate" (used a couple times above)
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1455,30 +1524,56 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
}
private void maybeGenerateSnapshot() {
- if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
- snapshotGeneratorManager.generator == null
- ) {
- if (!isActiveController()) {
- // The active controller creates in-memory snapshot every time an uncommitted
- // batch gets appended. The in-active controller can be more efficient and only
- // create an in-memory snapshot when needed.
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ if (!snapshotGeneratorManager.snapshotInProgress()) {
+ Set<SnapshotReason> snapshotReasons = new HashSet<>();
+ // Check if a snapshot should be generated because of committed bytes
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes) {
+ snapshotReasons.add(
+ SnapshotReason.maxBytesExceeded(committedBytesSinceLastSnapshot, snapshotMaxNewRecordBytes)
+ );
}
- log.info(
- "Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot because, {}.",
- lastCommittedEpoch,
- lastCommittedOffset,
- committedBytesSinceLastSnapshot,
- SnapshotReason.MaxBytesExceeded
- );
+ // Check if a snapshot should be generated because of committed append times
+ if (snapshotMaxIntervalMs > 0) {
+ // Time base snasphots are enabled
+ long snapshotIntervalMs = time.milliseconds() - oldestCommittedLogOnlyAppendTimestamp;
+ if (snapshotIntervalMs >= snapshotMaxIntervalMs) {
+ snapshotReasons.add(SnapshotReason.maxIntervalExceeded(snapshotIntervalMs, snapshotMaxIntervalMs));
+ } else {
+ maybeScheduleNextGenerateSnapshot();
+ }
+ }
- snapshotGeneratorManager.createSnapshotGenerator(
- lastCommittedOffset,
- lastCommittedEpoch,
- lastCommittedTimestamp
- );
- committedBytesSinceLastSnapshot = 0;
+ if (!snapshotReasons.isEmpty()) {
+ if (!isActiveController()) {
+ // The inactive controllers only create an in-memory snapshot when generating a snapshot. This is
+ // unlike the active controller which creates in-memory snapshots every time an uncommitted batch
+ // gets replayed.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
+
+ log.info(
+ "Generating a snapshot that includes (epoch={}, offset={}) because: {}",
+ lastCommittedEpoch,
+ lastCommittedOffset,
+ SnapshotReason.stringFromReasons(snapshotReasons)
+ );
+
+ snapshotGeneratorManager.createSnapshotGenerator(lastCommittedOffset, lastCommittedEpoch, lastCommittedTimestamp);
+
+ // Reset all of the snapshot counters
+ committedBytesSinceLastSnapshot = 0;
+ oldestCommittedLogOnlyAppendTimestamp = Long.MAX_VALUE;
+
+ // Starting a snapshot invalidates any scheduled snapshot generation
+ cancelNextGenerateSnapshot();
+ }
+ } else {
+ /* Skip snapshot generation if there is a snaphshot in progress.
Review Comment:
nit: I think this would be a little clearer if the `if` check is inverted:
```java
if (snapshotGeneratorManager.snapshotInProgress()) {
/* Skip snapshot generation if there is a snaphshot in progress.
...
} else {
```
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
+
+ // The snapshot counters for size-based and time-based snapshots could have changed to cause a new
+ // snapshot to get generated.
+ maybeGenerateSnapshot();
Review Comment:
In case there was a failure, does it make sense to back off before retrying?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
I do think `oldestNonSnapshottedLogAppendTimestamp` is a little clearer. Perhaps the fact that it is an append timestamp is already clear from context and we could use `oldestNonSnapshottedTimestamp`? I also don't feel strongly though.
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,15 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the time elapsed, see " +
+ s"the <code>$MetadataSnapshotMaxIntervalMsProp</code> configuration."
+ val MetadataSnapshotMaxIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
+ "if there are committed records in the log that are not included in the latest snapshot. A value of zero disables " +
+ s"time based snapshot generation. The default value is ${Defaults.MetadataSnapshotMaxIntervalMs}. To geneate " +
+ s"snapshots based on the number of metadata bytes, see the <code>$MetadataSnapshotMaxNewRecordBytesProp</code> " +
Review Comment:
Instead of just referring to the other configuration, I was thinking we could mention that snapshots will be taken when either the interval is reached or the max bytes limit is reached.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028467761
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
+
+ // The snapshot counters for size-based and time-based snapshots could have changed to cause a new
+ // snapshot to get generated.
+ maybeGenerateSnapshot();
Review Comment:
The quorum controller resets the size-based (`committedBytesSinceLastSnapshot`) and time-based (`oldestNonSnapshottedTimestamp`) variables when it starts a snapshot. If the snapshot fails these variables were reset then snapshot generation started.
This acts as a throttle. Snapshot generation will be trigger as often as what is described in `metadata.log.max.record.bytes.between.snapshots` and `metadata.log.max.snapshot.interval.ms` even if the snapshot happens to fail.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028471070
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,15 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the time elapsed, see " +
+ s"the <code>$MetadataSnapshotMaxIntervalMsProp</code> configuration."
+ val MetadataSnapshotMaxIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
+ "if there are committed records in the log that are not included in the latest snapshot. A value of zero disables " +
+ s"time based snapshot generation. The default value is ${Defaults.MetadataSnapshotMaxIntervalMs}. To geneate " +
+ s"snapshots based on the number of metadata bytes, see the <code>$MetadataSnapshotMaxNewRecordBytesProp</code> " +
Review Comment:
Done. Added a sentence to both descriptions explaining 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026780591
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,12 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ Defaults.MetadataSnapshotMaxNewRecordBytes
+ val MetadataMaxSnapshotIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
Review Comment:
Is it worth saying something about the relation between these two configs?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
+
+ // The snapshot counters for size-based and time-based snapshots could have changed to cause a new
+ // snapshot to get generated.
+ maybeGenerateSnapshot();
+ }
+
@Override
public void run() {
- if (generator == null) {
- log.debug("No snapshot is in progress.");
+ if (!inProgress()) {
+ log.debug("No snapshot is in progress because it was previously canceled");
return;
}
+
OptionalLong nextDelay;
try {
nextDelay = generator.generateBatches();
} catch (Exception e) {
- log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), e);
- generator.writer().close();
- generator = null;
+ handleSnapshotFinished(Optional.of(e));
return;
}
- if (!nextDelay.isPresent()) {
- log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
- generator.writer().close();
- generator = null;
- // Delete every in-memory snapshot up to the committed offset. They are not needed since this
- // snapshot generation finished.
- snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
- return;
+ if (nextDelay.isPresent()) {
+ reschedule(nextDelay.getAsLong());
+ } else {
+ handleSnapshotFinished(Optional.empty());
}
- reschedule(nextDelay.getAsLong());
}
OptionalLong snapshotLastOffsetFromLog() {
- if (generator == null) {
+ if (!inProgress()) {
return OptionalLong.empty();
}
return OptionalLong.of(generator.lastContainedLogOffset());
}
+
+ public boolean inProgress() {
Review Comment:
nit: maybe `snapshotInProgress` is a little cleaerer?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1455,30 +1523,56 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
}
private void maybeGenerateSnapshot() {
- if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
- snapshotGeneratorManager.generator == null
- ) {
- if (!isActiveController()) {
- // The active controller creates in-memory snapshot every time an uncommitted
- // batch gets appended. The in-active controller can be more efficient and only
- // create an in-memory snapshot when needed.
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ if (!snapshotGeneratorManager.inProgress()) {
+ Set<SnapshotReason> snapshotReasons = new HashSet<>();
+ // Check if a snapshot should be generated because of committed bytes
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes) {
+ snapshotReasons.add(
+ SnapshotReason.maxBytesExceeded(committedBytesSinceLastSnapshot, snapshotMaxNewRecordBytes)
+ );
}
- log.info(
- "Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot because, {}.",
- lastCommittedEpoch,
- lastCommittedOffset,
- committedBytesSinceLastSnapshot,
- SnapshotReason.MaxBytesExceeded
- );
+ // Check if a snapshot should be generated because of committed append times
+ if (maxSnapshotIntervalMs > 0) {
+ // Time base snasphots are enabled
+ long snapshotIntervalMs = time.milliseconds() - oldestCommittedLogOnlyAppendTimestamp;
+ if (snapshotIntervalMs >= maxSnapshotIntervalMs) {
+ snapshotReasons.add(SnapshotReason.maxIntervalExceeded(snapshotIntervalMs, maxSnapshotIntervalMs));
+ } else {
+ maybeScheduleNextGenerateSnapshot();
+ }
+ }
- snapshotGeneratorManager.createSnapshotGenerator(
- lastCommittedOffset,
- lastCommittedEpoch,
- lastCommittedTimestamp
- );
- committedBytesSinceLastSnapshot = 0;
+ if (!snapshotReasons.isEmpty()) {
+ if (!isActiveController()) {
+ // The active controller creates in-memory snapshot every time an uncommitted
+ // batch gets appended. The in-active controller can be more efficient and only
Review Comment:
nit: "inactive"?
##########
metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotReason.java:
##########
@@ -16,19 +16,48 @@
*/
package org.apache.kafka.metadata.util;
-public enum SnapshotReason {
- UnknownReason("unknown reason"),
- MaxBytesExceeded("max bytes were exceeded"),
- MetadataVersionChanged("metadata version was changed");
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import org.apache.kafka.server.common.MetadataVersion;
- private final String snapshotReason;
+/**
+ * Reason for generating a snapshot.
+ */
+public final class SnapshotReason {
+ static public final SnapshotReason UNKNOWN = new SnapshotReason("unknown reason");
+
+ static public SnapshotReason maxBytesExceeded(long bytes, long maxBytes) {
+ return new SnapshotReason(String.format("%s bytes exceeded the maximum bytes of %s", bytes, maxBytes));
+ }
+
+ static public SnapshotReason maxIntervalExceeded(long interval, long maxInterval) {
+ return new SnapshotReason(
+ String.format("%s ms exceeded the maximum snapshot interval of %s ms", interval, maxInterval)
+ );
+ }
+
+ static public SnapshotReason metadataVersionChanged(MetadataVersion metadataVersion) {
+ return new SnapshotReason(String.format("metadata version was changed to %s", metadataVersion));
+ }
+
+ /**
+ * Converts a collection of reasons into a string.
+ */
+ static public String stringFromReasons(Iterable<SnapshotReason> reasons) {
Review Comment:
I think you can use `Utils.join` as well.
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1157,6 +1164,7 @@ object KafkaConfig {
* KRaft mode configs.
*/
.define(MetadataSnapshotMaxNewRecordBytesProp, LONG, Defaults.MetadataSnapshotMaxNewRecordBytes, atLeast(1), HIGH, MetadataSnapshotMaxNewRecordBytesDoc)
+ .define(MetadataMaxSnapshotIntervalMsProp, LONG, Defaults.MetadataMaxSnapshotIntervalMs, atLeast(0), HIGH, MetadataMaxSnapshotIntervalMsDoc)
Review Comment:
What is the expected behavior if the value is 0?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
If there is a snapshot failure, will we wait for the next interval before retrying? I wonder if we need a metric or something to get marked when this happens?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
Why "LogOnlyAppendTimestamp" instead of "LogAppendTimestamp"?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026995028
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
Doing it separately sounds fine.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026974404
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,12 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ Defaults.MetadataSnapshotMaxNewRecordBytes
+ val MetadataMaxSnapshotIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
Review Comment:
Updated the documentation. Let me know what you think.
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
+
+ // The snapshot counters for size-based and time-based snapshots could have changed to cause a new
+ // snapshot to get generated.
+ maybeGenerateSnapshot();
+ }
+
@Override
public void run() {
- if (generator == null) {
- log.debug("No snapshot is in progress.");
+ if (!inProgress()) {
+ log.debug("No snapshot is in progress because it was previously canceled");
return;
}
+
OptionalLong nextDelay;
try {
nextDelay = generator.generateBatches();
} catch (Exception e) {
- log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), e);
- generator.writer().close();
- generator = null;
+ handleSnapshotFinished(Optional.of(e));
return;
}
- if (!nextDelay.isPresent()) {
- log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
- generator.writer().close();
- generator = null;
- // Delete every in-memory snapshot up to the committed offset. They are not needed since this
- // snapshot generation finished.
- snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
- return;
+ if (nextDelay.isPresent()) {
+ reschedule(nextDelay.getAsLong());
+ } else {
+ handleSnapshotFinished(Optional.empty());
}
- reschedule(nextDelay.getAsLong());
}
OptionalLong snapshotLastOffsetFromLog() {
- if (generator == null) {
+ if (!inProgress()) {
return OptionalLong.empty();
}
return OptionalLong.of(generator.lastContainedLogOffset());
}
+
+ public boolean inProgress() {
Review Comment:
Done.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026950932
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1157,6 +1164,7 @@ object KafkaConfig {
* KRaft mode configs.
*/
.define(MetadataSnapshotMaxNewRecordBytesProp, LONG, Defaults.MetadataSnapshotMaxNewRecordBytes, atLeast(1), HIGH, MetadataSnapshotMaxNewRecordBytesDoc)
+ .define(MetadataMaxSnapshotIntervalMsProp, LONG, Defaults.MetadataMaxSnapshotIntervalMs, atLeast(0), HIGH, MetadataMaxSnapshotIntervalMsDoc)
Review Comment:
A value of zero disables time based snapshots. I have this wording in the properties description:
> A value of zero disables time based snapshot generation.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028472130
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
I picked `oldestNonSnapshottedTimestamp`. Updated the PR.l
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027024757
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1376,6 +1413,37 @@ private void cancelNextWriteNoOpRecord() {
queue.cancelDeferred(WRITE_NO_OP_RECORD);
}
+ private static final String MAYBE_GENERATE_SNAPSHOT = "maybeGenerateSnapshot";
+
+ private void maybeScheduleNextGenerateSnapshot() {
+ if (!generateSnapshotScheduled) {
+ long now = time.milliseconds();
+ long delayMs = Math.min(
+ 0,
+ maxSnapshotIntervalMs + oldestCommittedLogOnlyAppendTimestamp - now
+ );
+
+ log.debug(
+ "Scheduling write event for {} because maxSnapshotIntervalMs ({}), " +
+ "oldestCommittedLogOnlyAppendTimestamp ({}) and now ({})",
+ maxSnapshotIntervalMs,
Review Comment:
Yes. Good catch. Fixed.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026957838
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
> If there is a snapshot failure, will we wait for the next interval before retrying?
Yeah. The snapshot counters `committedBytesSinceLastSnapshot` and `oldestCommittedLogOnlyAppendTimestamp` are updated assuming that snapshot generation cannot fail.
We could add another reason ("snapshot failure") that triggers a snapshot immediately irrespective of the counters. I am concerned that this event/task my starve other controller events. Maybe it is better to just rely on timed snapshot and NoOpRecord to trigger another snapshot in the case of failures.
> If there is a snapshot failure, will we wait for the next interval before retrying?
Let me check the KIPs but I don't think we define this metric. I can add it and update one of the KIPs if we agree to it 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027022584
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
Review Comment:
Yes. The controller needs to keep in-memory snapshot older than the last committed offset because a snapshot may be iterating over those timelined values. Once we know that there are no pending snapshots (either because it succeeded or it failed) the controller can delete in-memory snapshots up to the committed offset.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028467761
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
+
+ // The snapshot counters for size-based and time-based snapshots could have changed to cause a new
+ // snapshot to get generated.
+ maybeGenerateSnapshot();
Review Comment:
The quorum controller resets the size-based (`committedBytesSinceLastSnapshot`) and time-based (`oldestNonSnapshottedTimestamp`) variables when it starts a snapshot. If the snapshot fails these variables were still reset when snapshot generation started.
This acts as a throttle. Snapshot generation will be trigger as often as what is described in `metadata.log.max.record.bytes.between.snapshots` and `metadata.log.max.snapshot.interval.ms` even if the snapshot happens to fail.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio merged pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio merged PR #12761:
URL: https://github.com/apache/kafka/pull/12761
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027024702
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,14 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the number of metadata byets see " +
Review Comment:
Yes. Fixed.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r999705130
##########
raft/src/main/java/org/apache/kafka/raft/OffsetAndEpoch.java:
##########
@@ -25,6 +25,14 @@ public OffsetAndEpoch(long offset, int epoch) {
this.epoch = epoch;
}
+ public long offset() {
+ return offset;
+ }
+
+ public int epoch() {
Review Comment:
@hachikuji https://github.com/apache/kafka/pull/12770
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028471411
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,15 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the time elapsed, see " +
+ s"the <code>$MetadataSnapshotMaxIntervalMsProp</code> configuration."
+ val MetadataSnapshotMaxIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
+ "if there are committed records in the log that are not included in the latest snapshot. A value of zero disables " +
+ s"time based snapshot generation. The default value is ${Defaults.MetadataSnapshotMaxIntervalMs}. To geneate " +
Review Comment:
Thanks. Done. I need a spell checker in my editor.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027019732
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
Okay. Let's move this discussion to the DISCUSSION thread in Kafka dev. I'll send a message next week.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027024629
##########
raft/src/main/java/org/apache/kafka/raft/RaftClient.java:
##########
@@ -229,4 +229,15 @@ default void beginShutdown() {}
* or less than the log start offset.
*/
Optional<SnapshotWriter<T>> createSnapshot(long committedOffset, int committedEpoch, long lastContainedLogTime);
+
+
+ /**
+ * The snapshot id for the lastest snapshot.
+ *
+ * Returns the snapshot id of the latest snapshot, if it exists. If a snapshot doesn't exists, returns an
+ * {@link Optional#empty()}.
+ *
+ * @return the snapshot of the latest snaphost, if it exists
Review Comment:
Done.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026965577
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
It is the oldest timestamp in the log that is not included in a snapshot. I was thinking that `oldestCommittedLogAppendTimestamp` could mislead the reader if they don't read the description of that variable:
```java
/**
* Timestamp for the oldest record that was committed but not included in a snapshot.
*/
private long oldestCommittedLogOnlyAppendTimestamp = Long.MAX_VALUE;
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
niket-goel commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027000889
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
> kafka.controller:type=KafkaController,name=MetadataSnapshotGenerationErrors
I am concerned that with a counter that resets at restart, we might have an ever so small likelihood of never incrementing the metric even with the log moving and no snapshots generated e.g. if we keep crashing while generating a snapshot. Ditto for the broker.
What do you think about having a `kafka.controller:type=KafkaController,name=LastSnapshotTime` metric.
This metric would just emit the timestamp (linux epoch) of when the last snapshot was successfully written.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026974492
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1455,30 +1523,56 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
}
private void maybeGenerateSnapshot() {
- if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
- snapshotGeneratorManager.generator == null
- ) {
- if (!isActiveController()) {
- // The active controller creates in-memory snapshot every time an uncommitted
- // batch gets appended. The in-active controller can be more efficient and only
- // create an in-memory snapshot when needed.
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ if (!snapshotGeneratorManager.inProgress()) {
+ Set<SnapshotReason> snapshotReasons = new HashSet<>();
+ // Check if a snapshot should be generated because of committed bytes
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes) {
+ snapshotReasons.add(
+ SnapshotReason.maxBytesExceeded(committedBytesSinceLastSnapshot, snapshotMaxNewRecordBytes)
+ );
}
- log.info(
- "Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot because, {}.",
- lastCommittedEpoch,
- lastCommittedOffset,
- committedBytesSinceLastSnapshot,
- SnapshotReason.MaxBytesExceeded
- );
+ // Check if a snapshot should be generated because of committed append times
+ if (maxSnapshotIntervalMs > 0) {
+ // Time base snasphots are enabled
+ long snapshotIntervalMs = time.milliseconds() - oldestCommittedLogOnlyAppendTimestamp;
+ if (snapshotIntervalMs >= maxSnapshotIntervalMs) {
+ snapshotReasons.add(SnapshotReason.maxIntervalExceeded(snapshotIntervalMs, maxSnapshotIntervalMs));
+ } else {
+ maybeScheduleNextGenerateSnapshot();
+ }
+ }
- snapshotGeneratorManager.createSnapshotGenerator(
- lastCommittedOffset,
- lastCommittedEpoch,
- lastCommittedTimestamp
- );
- committedBytesSinceLastSnapshot = 0;
+ if (!snapshotReasons.isEmpty()) {
+ if (!isActiveController()) {
+ // The active controller creates in-memory snapshot every time an uncommitted
+ // batch gets appended. The in-active controller can be more efficient and only
Review Comment:
Done.
##########
metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotReason.java:
##########
@@ -16,19 +16,48 @@
*/
package org.apache.kafka.metadata.util;
-public enum SnapshotReason {
- UnknownReason("unknown reason"),
- MaxBytesExceeded("max bytes were exceeded"),
- MetadataVersionChanged("metadata version was changed");
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import org.apache.kafka.server.common.MetadataVersion;
- private final String snapshotReason;
+/**
+ * Reason for generating a snapshot.
+ */
+public final class SnapshotReason {
+ static public final SnapshotReason UNKNOWN = new SnapshotReason("unknown reason");
+
+ static public SnapshotReason maxBytesExceeded(long bytes, long maxBytes) {
+ return new SnapshotReason(String.format("%s bytes exceeded the maximum bytes of %s", bytes, maxBytes));
+ }
+
+ static public SnapshotReason maxIntervalExceeded(long interval, long maxInterval) {
+ return new SnapshotReason(
+ String.format("%s ms exceeded the maximum snapshot interval of %s ms", interval, maxInterval)
+ );
+ }
+
+ static public SnapshotReason metadataVersionChanged(MetadataVersion metadataVersion) {
+ return new SnapshotReason(String.format("metadata version was changed to %s", metadataVersion));
+ }
+
+ /**
+ * Converts a collection of reasons into a string.
+ */
+ static public String stringFromReasons(Iterable<SnapshotReason> reasons) {
Review Comment:
Done.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028514280
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1455,30 +1524,56 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
}
private void maybeGenerateSnapshot() {
- if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
- snapshotGeneratorManager.generator == null
- ) {
- if (!isActiveController()) {
- // The active controller creates in-memory snapshot every time an uncommitted
- // batch gets appended. The in-active controller can be more efficient and only
- // create an in-memory snapshot when needed.
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ if (!snapshotGeneratorManager.snapshotInProgress()) {
+ Set<SnapshotReason> snapshotReasons = new HashSet<>();
+ // Check if a snapshot should be generated because of committed bytes
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes) {
+ snapshotReasons.add(
+ SnapshotReason.maxBytesExceeded(committedBytesSinceLastSnapshot, snapshotMaxNewRecordBytes)
+ );
}
- log.info(
- "Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot because, {}.",
- lastCommittedEpoch,
- lastCommittedOffset,
- committedBytesSinceLastSnapshot,
- SnapshotReason.MaxBytesExceeded
- );
+ // Check if a snapshot should be generated because of committed append times
+ if (snapshotMaxIntervalMs > 0) {
+ // Time base snasphots are enabled
+ long snapshotIntervalMs = time.milliseconds() - oldestCommittedLogOnlyAppendTimestamp;
+ if (snapshotIntervalMs >= snapshotMaxIntervalMs) {
+ snapshotReasons.add(SnapshotReason.maxIntervalExceeded(snapshotIntervalMs, snapshotMaxIntervalMs));
+ } else {
+ maybeScheduleNextGenerateSnapshot();
+ }
+ }
- snapshotGeneratorManager.createSnapshotGenerator(
- lastCommittedOffset,
- lastCommittedEpoch,
- lastCommittedTimestamp
- );
- committedBytesSinceLastSnapshot = 0;
+ if (!snapshotReasons.isEmpty()) {
+ if (!isActiveController()) {
+ // The inactive controllers only create an in-memory snapshot when generating a snapshot. This is
+ // unlike the active controller which creates in-memory snapshots every time an uncommitted batch
+ // gets replayed.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
+
+ log.info(
+ "Generating a snapshot that includes (epoch={}, offset={}) because: {}",
+ lastCommittedEpoch,
+ lastCommittedOffset,
+ SnapshotReason.stringFromReasons(snapshotReasons)
+ );
+
+ snapshotGeneratorManager.createSnapshotGenerator(lastCommittedOffset, lastCommittedEpoch, lastCommittedTimestamp);
+
+ // Reset all of the snapshot counters
+ committedBytesSinceLastSnapshot = 0;
+ oldestCommittedLogOnlyAppendTimestamp = Long.MAX_VALUE;
+
+ // Starting a snapshot invalidates any scheduled snapshot generation
+ cancelNextGenerateSnapshot();
+ }
+ } else {
+ /* Skip snapshot generation if there is a snaphshot in progress.
Review Comment:
Done.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1028471411
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,15 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the time elapsed, see " +
+ s"the <code>$MetadataSnapshotMaxIntervalMsProp</code> configuration."
+ val MetadataSnapshotMaxIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
+ "if there are committed records in the log that are not included in the latest snapshot. A value of zero disables " +
+ s"time based snapshot generation. The default value is ${Defaults.MetadataSnapshotMaxIntervalMs}. To geneate " +
Review Comment:
Thanks. Done. I need a spell check in my editor.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r999563515
##########
raft/src/main/java/org/apache/kafka/raft/OffsetAndEpoch.java:
##########
@@ -25,6 +25,14 @@ public OffsetAndEpoch(long offset, int epoch) {
this.epoch = epoch;
}
+ public long offset() {
+ return offset;
+ }
+
+ public int epoch() {
Review Comment:
Yes. Do you mind if I do this in another PR? I don't want to add too many trivial diffs to this one.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026963407
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
How about this for the controller:
`kafka.controller:type=KafkaController,name=MetadataSnapshotGenerationErrors`
Incremented anytime the controller fails to generate a snapshot. Reset to zero anytime the controller restarts or a snapshot is successfully generated.
And this for the brokers:
kafka.server:type=broker-metadata-metrics,name=snapshot-generation-errors
Incremented anytime the broker fails to generate a snapshot. Reset to zero anytime the broker restarts or a snapshot is successfully generated.
@hachikuji do you mind if I implement this in a future PR? I created this issue: https://issues.apache.org/jira/browse/KAFKA-14403
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
niket-goel commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1026985550
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,14 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the number of metadata byets see " +
+ s"the <code>$MetadataMaxSnapshotIntervalMsProp</code> configuration."
+ val MetadataMaxSnapshotIntervalMsDoc = "This is the maximum number of milliseconds to wait to generate a snapshot " +
+ "if there are records in the log that are not included in the latest snapshot. A value of zero disables time based " +
+ s"snapshot generation. The default value is ${Defaults.MetadataMaxSnapshotIntervalMs}. To geneate snapshots based " +
+ s"on the number of metadata byets see the <code>$MetadataSnapshotMaxNewRecordBytesProp</code> configuration."
Review Comment:
Nit: typo in "bytes"
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -81,6 +81,7 @@ object Defaults {
val BrokerHeartbeatIntervalMs = 2000
val BrokerSessionTimeoutMs = 9000
val MetadataSnapshotMaxNewRecordBytes = 20 * 1024 * 1024
+ val MetadataMaxSnapshotIntervalMs = TimeUnit.HOURS.toMillis(1);
Review Comment:
Ditto on naming.
##########
raft/src/main/java/org/apache/kafka/raft/RaftClient.java:
##########
@@ -229,4 +229,15 @@ default void beginShutdown() {}
* or less than the log start offset.
*/
Optional<SnapshotWriter<T>> createSnapshot(long committedOffset, int committedEpoch, long lastContainedLogTime);
+
+
+ /**
+ * The snapshot id for the lastest snapshot.
+ *
+ * Returns the snapshot id of the latest snapshot, if it exists. If a snapshot doesn't exists, returns an
+ * {@link Optional#empty()}.
+ *
+ * @return the snapshot of the latest snaphost, if it exists
Review Comment:
%s/the snapshot of/ the id of/
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -725,7 +727,14 @@ object KafkaConfig {
"This is required configuration when running in KRaft mode."
val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
"If it is not set, the metadata log is placed in the first log directory from log.dirs."
- val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+ val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest " +
+ "snapshot and the high-watermark needed before generating a new snapshot. The default value is " +
+ s"${Defaults.MetadataSnapshotMaxNewRecordBytes}. To geneate snapshots based on the number of metadata byets see " +
Review Comment:
Did you mean to say _"To generate snapshots based on elapsed time"_?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1376,6 +1413,37 @@ private void cancelNextWriteNoOpRecord() {
queue.cancelDeferred(WRITE_NO_OP_RECORD);
}
+ private static final String MAYBE_GENERATE_SNAPSHOT = "maybeGenerateSnapshot";
+
+ private void maybeScheduleNextGenerateSnapshot() {
+ if (!generateSnapshotScheduled) {
+ long now = time.milliseconds();
+ long delayMs = Math.min(
+ 0,
+ maxSnapshotIntervalMs + oldestCommittedLogOnlyAppendTimestamp - now
+ );
+
+ log.debug(
+ "Scheduling write event for {} because maxSnapshotIntervalMs ({}), " +
+ "oldestCommittedLogOnlyAppendTimestamp ({}) and now ({})",
+ maxSnapshotIntervalMs,
Review Comment:
Are we missing an argument?
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
+ } else {
+ log.info("Finished generating snapshot {}.", generator.lastContainedLogOffset());
+ }
+
+ generator.writer().close();
+ generator = null;
+
+ // Delete every in-memory snapshot up to the committed offset. They are not needed since this
+ // snapshot generation finished.
+ snapshotRegistry.deleteSnapshotsUpTo(lastCommittedOffset);
Review Comment:
Question: Is this OK to do even if the snapshot never succeeded? I think the answer would be yes, but just confirming.
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
nit: Maybe overkill, but how about something even more verbose like `oldestNonSnapshottedLogAppendTimestamp`.
The current name still had me read the description.
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1455,30 +1523,56 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
}
private void maybeGenerateSnapshot() {
- if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
- snapshotGeneratorManager.generator == null
- ) {
- if (!isActiveController()) {
- // The active controller creates in-memory snapshot every time an uncommitted
- // batch gets appended. The in-active controller can be more efficient and only
- // create an in-memory snapshot when needed.
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ if (!snapshotGeneratorManager.snapshotInProgress()) {
+ Set<SnapshotReason> snapshotReasons = new HashSet<>();
+ // Check if a snapshot should be generated because of committed bytes
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes) {
+ snapshotReasons.add(
+ SnapshotReason.maxBytesExceeded(committedBytesSinceLastSnapshot, snapshotMaxNewRecordBytes)
+ );
}
- log.info(
- "Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot because, {}.",
- lastCommittedEpoch,
- lastCommittedOffset,
- committedBytesSinceLastSnapshot,
- SnapshotReason.MaxBytesExceeded
- );
+ // Check if a snapshot should be generated because of committed append times
+ if (maxSnapshotIntervalMs > 0) {
+ // Time base snasphots are enabled
+ long snapshotIntervalMs = time.milliseconds() - oldestCommittedLogOnlyAppendTimestamp;
+ if (snapshotIntervalMs >= maxSnapshotIntervalMs) {
+ snapshotReasons.add(SnapshotReason.maxIntervalExceeded(snapshotIntervalMs, maxSnapshotIntervalMs));
+ } else {
+ maybeScheduleNextGenerateSnapshot();
+ }
+ }
- snapshotGeneratorManager.createSnapshotGenerator(
- lastCommittedOffset,
- lastCommittedEpoch,
- lastCommittedTimestamp
- );
- committedBytesSinceLastSnapshot = 0;
+ if (!snapshotReasons.isEmpty()) {
+ if (!isActiveController()) {
+ // The active controller creates in-memory snapshot every time an uncommitted
Review Comment:
Nit:
Would be a little easier to read if the comment was rephrased to put what is happening here:
```
// The standby controllers do not create in-memory snapshots every time a
// batch gets appended, so we create it now.
```
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -201,6 +201,7 @@ class ControllerServer(
setSessionTimeoutNs(TimeUnit.NANOSECONDS.convert(config.brokerSessionTimeoutMs.longValue(),
TimeUnit.MILLISECONDS)).
setSnapshotMaxNewRecordBytes(config.metadataSnapshotMaxNewRecordBytes).
+ setMaxSnapshotIntervalMs(config.metadataMaxSnapshotIntervalMs).
Review Comment:
Nit: Better to be consistent with naming, e.g. `setSnapshotMaxIntervalMs` to align with `setSnapshotMaxNewRecordBytes`
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] niket-goel commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
niket-goel commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027000889
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -568,45 +576,62 @@ void cancel() {
}
void reschedule(long delayNs) {
- ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this);
+ ControllerEvent event = new ControllerEvent(GENERATE_SNAPSHOT, this);
queue.scheduleDeferred(event.name,
new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event);
}
+ void handleSnapshotFinished(Optional<Exception> exception) {
+ if (exception.isPresent()) {
+ log.error("Error while generating snapshot {}", generator.lastContainedLogOffset(), exception.get());
Review Comment:
> kafka.controller:type=KafkaController,name=MetadataSnapshotGenerationErrors
I am concerned that with a counter that resets at restart, we might have a ever so small likelihood of never incrementing the metric even with the log moving if we e.g. keep crashing while generating a snapshot. Ditto for the broker.
What do you think about having a `kafka.controller:type=KafkaController,name=LastSnapshotTime` metric.
This metric would just emit the timestamp (linux epoch) of when the last snapshot was successfully written.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027026027
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1455,30 +1523,56 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
}
private void maybeGenerateSnapshot() {
- if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes &&
- snapshotGeneratorManager.generator == null
- ) {
- if (!isActiveController()) {
- // The active controller creates in-memory snapshot every time an uncommitted
- // batch gets appended. The in-active controller can be more efficient and only
- // create an in-memory snapshot when needed.
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ if (!snapshotGeneratorManager.snapshotInProgress()) {
+ Set<SnapshotReason> snapshotReasons = new HashSet<>();
+ // Check if a snapshot should be generated because of committed bytes
+ if (committedBytesSinceLastSnapshot >= snapshotMaxNewRecordBytes) {
+ snapshotReasons.add(
+ SnapshotReason.maxBytesExceeded(committedBytesSinceLastSnapshot, snapshotMaxNewRecordBytes)
+ );
}
- log.info(
- "Generating a snapshot that includes (epoch={}, offset={}) after {} committed bytes since the last snapshot because, {}.",
- lastCommittedEpoch,
- lastCommittedOffset,
- committedBytesSinceLastSnapshot,
- SnapshotReason.MaxBytesExceeded
- );
+ // Check if a snapshot should be generated because of committed append times
+ if (maxSnapshotIntervalMs > 0) {
+ // Time base snasphots are enabled
+ long snapshotIntervalMs = time.milliseconds() - oldestCommittedLogOnlyAppendTimestamp;
+ if (snapshotIntervalMs >= maxSnapshotIntervalMs) {
+ snapshotReasons.add(SnapshotReason.maxIntervalExceeded(snapshotIntervalMs, maxSnapshotIntervalMs));
+ } else {
+ maybeScheduleNextGenerateSnapshot();
+ }
+ }
- snapshotGeneratorManager.createSnapshotGenerator(
- lastCommittedOffset,
- lastCommittedEpoch,
- lastCommittedTimestamp
- );
- committedBytesSinceLastSnapshot = 0;
+ if (!snapshotReasons.isEmpty()) {
+ if (!isActiveController()) {
+ // The active controller creates in-memory snapshot every time an uncommitted
Review Comment:
I reorder the sentences and improved the wording. I kept both sentences because I think they are important for anyone interested in understanding when the controllers generates and deletes in-memory snapshots.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12761: KAFKA-14307; Controller time-based snapshots
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12761:
URL: https://github.com/apache/kafka/pull/12761#discussion_r1027024376
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -986,6 +1012,13 @@ public void handleCommit(BatchReader<ApiMessageAndVersion> reader) {
batch.appendTimestamp(),
committedBytesSinceLastSnapshot + batch.sizeInBytes()
);
+
+ if (offset >= raftClient.latestSnapshotId().map(OffsetAndEpoch::offset).orElse(0L)) {
+ oldestCommittedLogOnlyAppendTimestamp = Math.min(
Review Comment:
I was thinking of a similar name when I was implementing this feature but I felt it was too verbose. I don't have a strong opinion here so I can change it to whatever the reader prefers.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org