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/06/08 07:11:53 UTC
[GitHub] [kafka] dengziming opened a new pull request, #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
dengziming opened a new pull request, #12265:
URL: https://github.com/apache/kafka/pull/12265
*More detailed description of your change*
There are 3 bugs when a broker generates a snapshot.
1. Broker should not generate snapshots until it starts publishing.
Before a broker starts publishing, `BrokerMetadataListener._publisher=None`, so `_publisher.foreach(publish)` will do nothing, so `featuresDelta.metadataVersionChange().isPresent` is always true, so we will generating a snapshot on every commit, here are the logs, note offset 1 is a LeaderChangeMessage so there is no snapshot:
```
[2022-06-08 13:07:43,010] INFO [BrokerMetadataSnapshotter id=0] Creating a new snapshot at offset 0... (kafka.server.metadata.BrokerMetadataSnapshotter:66)
[2022-06-08 13:07:43,222] INFO [BrokerMetadataSnapshotter id=0] Creating a new snapshot at offset 2... (kafka.server.metadata.BrokerMetadataSnapshotter:66)
[2022-06-08 13:07:43,727] INFO [BrokerMetadataSnapshotter id=0] Creating a new snapshot at offset 3... (kafka.server.metadata.BrokerMetadataSnapshotter:66)
[2022-06-08 13:07:44,228] INFO [BrokerMetadataSnapshotter id=0] Creating a new snapshot at offset 4... (kafka.server.metadata.BrokerMetadataSnapshotter:66)
```
2. We should compute `metadataVersionChanged` before `_publisher.foreach(publish)`
After `_publisher.foreach(publish)` the `BrokerMetadataListener_delta` is always Empty, so `metadataVersionChanged` is always false, this means we will never trigger snapshot generating due to metadata version change.
3. We should try to generate a snapshot when starting publishing
When we started publishing, there may be a metadata version change, so we should try to generate a snapshot before first publishing.
*Summary of testing strategy (including rationale)*
A unit test.
### 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] hachikuji commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r914289201
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,26 +45,37 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
+ /**
+ * The offset of the newest snapshot, or -1 if there hasn't been one.Accessed only under
+ * the object lock.
+ */
+ private var _latestSnapshotOffset = -1L
+
/**
* The event queue which runs this listener.
*/
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
- if (_currentSnapshotOffset == -1L) {
+ if (_currentSnapshotOffset != -1) {
+ warn(s"Declining to create a new snapshot at ${image.highestOffsetAndEpoch()} because " +
+ s"there is already a snapshot in progress at offset ${_currentSnapshotOffset}")
+ false
+ } else if (_latestSnapshotOffset >= image.highestOffsetAndEpoch().offset) {
Review Comment:
Is the test failure hitting the case when `_latestSnapshotOffset` is equal to the `image.highestOffsetAndEpoch().offset`? Unless we have lost committed data, there shouldn't be any situation where the image offset is less than our snapshot 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 merged pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
jsancio merged PR #12265:
URL: https://github.com/apache/kafka/pull/12265
--
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] divijvaidya commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r908307779
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(100L, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV2.featureLevel(), 100L)
Review Comment:
This test would fail when the test is running with MetadataVersion.IBP_3_3_IV2 for some reason. Perhaps, add an assertion in the test
```
assertTrue(metadataVersion.isAtLeast(IBP_3_3_IV3))
```
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
Review Comment:
nit
refactor out endoffset into a variable and use the same variable for assertion
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
Review Comment:
Amongst all these tests we want to ensure that a stray `handleSnaphot` event does not modify the logic we expected. Perhaps spy the MetadataListener and assert that no calls to `handleSnapshot` was made. This would ensure that the snaphot was triggered specifically because of version change and not because of some external event.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r914533237
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,26 +45,37 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
+ /**
+ * The offset of the newest snapshot, or -1 if there hasn't been one.Accessed only under
+ * the object lock.
+ */
+ private var _latestSnapshotOffset = -1L
+
/**
* The event queue which runs this listener.
*/
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
- if (_currentSnapshotOffset == -1L) {
+ if (_currentSnapshotOffset != -1) {
+ warn(s"Declining to create a new snapshot at ${image.highestOffsetAndEpoch()} because " +
+ s"there is already a snapshot in progress at offset ${_currentSnapshotOffset}")
+ false
+ } else if (_latestSnapshotOffset >= image.highestOffsetAndEpoch().offset) {
Review Comment:
Yes, the test failed when generating a snapshot twice at the same offset firstly due to enough bytes having accumulated and secondly due to the metadata version changed. I changed it to "==" to make it more accurate and added a unit test for it.
--
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 #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r907665218
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
Review Comment:
Perhaps we could move the publisher check into `metadataVersionChanged()`? My understanding is that `metadataVersionChanged` will always returns true until we have applied the first delta, so that would make the dependence a little clearer. Do I have that right?
A consequence of changing that is that we could publish snapshots before the publisher has started if enough bytes have accumulated, which was the prior behavior. That would mean that we wouldn't need to update `StartPublishingEvent` below. Do you see any issues with that?
--
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] dengziming commented on pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on PR #12265:
URL: https://github.com/apache/kafka/pull/12265#issuecomment-1170181952
@hachikuji The cause of`RaftClusterSnapshotTest.testSnapshotsGenerated` has been found and the other is still under investigation.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r908207867
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
Review Comment:
This suggestion makes the logic clearer, but we still have to update `StartPublishingEvent` since there may be a `metadataVersionChanged`.
--
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] divijvaidya commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r892101409
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
+ maybeStartSnapshot()
}
- snapshotter.foreach { snapshotter =>
- _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot() || metadataVersionChanged) {
- if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
- _bytesSinceLastSnapshot = 0L
- }
- }
- }
+ _publisher.foreach(publish)
}
}
private def shouldSnapshot(): Boolean = {
- _bytesSinceLastSnapshot >= maxBytesBetweenSnapshots
+ _bytesSinceLastSnapshot >= maxBytesBetweenSnapshots || metadataVersionChanged()
Review Comment:
nit
personally I prefer to use parentheses for explicit grouping so that reader doesn't have to guess the precedence order amongst binary operators. It also avoids inadvertent bugs.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
+ maybeStartSnapshot()
}
- snapshotter.foreach { snapshotter =>
- _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot() || metadataVersionChanged) {
- if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
- _bytesSinceLastSnapshot = 0L
- }
- }
- }
+ _publisher.foreach(publish)
Review Comment:
could we rename `publish` to `publishAndResetDelta`? This will make it very implicit that the publish method resets the state set by loadBatches.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -244,6 +248,9 @@ class BrokerMetadataListener(
_publisher = Some(publisher)
log.info(s"Starting to publish metadata events at offset $highestMetadataOffset.")
try {
+ if (shouldSnapshot()) {
Review Comment:
For my better understanding here, please let me know if my assumptions are correct:
A listener can potentially receive a `handleCommit` or a `handleSnapshot` even before it has started publishing. It is because there is a duration of time between when a metadataListener is registered and a metaListener starts publishing. Yes?
In such cases (when listener has not started publishing) `handleCommit` will create a snapshot if it detects a metadata version change.
Then why do we need to create a snapshot on start publishing? Isn't it guaranteed that there would have been a `handleCommit` prior to this which would already have created the snapshot?
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +244,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(100L, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
Review Comment:
could we also add a negative test here which would have failed before we fixed the bug where publish was resetting _delta and hence we were doing a snapshot on every commit. Perhaps a test that validates that the snapshot is not occurring more than expected.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r915514842
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,26 +45,37 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
+ /**
+ * The offset of the newest snapshot, or -1 if there hasn't been one. Accessed only under
+ * the object lock.
+ */
+ private var _latestSnapshotOffset = -1L
+
/**
* The event queue which runs this listener.
*/
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
- if (_currentSnapshotOffset == -1L) {
+ if (_currentSnapshotOffset != -1) {
+ warn(s"Declining to create a new snapshot at ${image.highestOffsetAndEpoch()} because " +
Review Comment:
I agree with you here, info level is more suitable.
--
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] dengziming commented on pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on PR #12265:
URL: https://github.com/apache/kafka/pull/12265#issuecomment-1182684922
KAFKA-14073 is created for log improvement.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r895799683
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -244,6 +248,9 @@ class BrokerMetadataListener(
_publisher = Some(publisher)
log.info(s"Starting to publish metadata events at offset $highestMetadataOffset.")
try {
+ if (shouldSnapshot()) {
Review Comment:
1. Yes, a listener(here we refer to BrokerMetadataListener) will potentially receive a handleCommit or a handleSnapshot every time when it receives a metadata log change(for example, create a topic), when a broker startup, it will not start publishing until it gets all the metadata logs(I call this period "Catching up period" because a broker needs to connect to the controller and fetch all metadata logs).
2. We used to create a snapshot if it detects a metadata version change or _bytesSinceLastSnapshot is bigger enough.
3. But I changed the logic here to not create a snapshot in "Catching up period", so when starting publishing, which means the end of "Catching up period", we should inspect that is it necessary to create a snapshot.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
+ maybeStartSnapshot()
}
- snapshotter.foreach { snapshotter =>
- _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot() || metadataVersionChanged) {
- if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
- _bytesSinceLastSnapshot = 0L
- }
- }
- }
+ _publisher.foreach(publish)
Review Comment:
I think changing the name is a little overkill here, we may do many other things besides resetting delta.
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +244,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(100L, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
Review Comment:
This seems tested by `testNotSnapshotBeforePublishing` or I forgot about it, I will recheck it later.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r914535136
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(100L, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV2.featureLevel(), 100L)
Review Comment:
You are right, here I tried a different solution to update the feature level to `MetadataVersion.latest().featureLevel() - 1`, then we can be sure it's different from the current feature level, PTAL.
--
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 #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r915992223
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,32 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (shouldSnapshot()) {
+ maybeStartSnapshot()
}
- snapshotter.foreach { snapshotter =>
- _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot() || metadataVersionChanged) {
- if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
- _bytesSinceLastSnapshot = 0L
- }
- }
- }
+ _publisher.foreach(publish)
}
}
private def shouldSnapshot(): Boolean = {
- _bytesSinceLastSnapshot >= maxBytesBetweenSnapshots
+ (_bytesSinceLastSnapshot >= maxBytesBetweenSnapshots) || metadataVersionChanged()
+ }
+
+ private def metadataVersionChanged(): Boolean = {
+ _publisher.nonEmpty && Option(_delta.featuresDelta()).exists { featuresDelta =>
Review Comment:
Do you mind writing a comment as to why we check that the publisher is set. If I understand this correctly it is not a correctness issue but a performance issue, right? If I remember correctly, @mumrah mentioned that he wanted to generate a snapshot whenever the metadata version changes. Unfortunately, I couldn't find a mention of this in KIP-778.
With this change this is no longer true. What do you think @mumrah ?
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,42 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot on metadata version change before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ val endOffset = 100L
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), endOffset)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(endOffset, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+
+ val endOffset = 100L
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, (MetadataVersion.latest().featureLevel() - 1).toShort, endOffset)
+ listener.getImageRecords().get()
Review Comment:
Let's write a comment saying that this `get` is waiting for the metadata version update above to get processed.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -51,20 +51,26 @@ class BrokerMetadataSnapshotter(
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
- if (_currentSnapshotOffset == -1L) {
+ if (_currentSnapshotOffset != -1) {
+ info(s"Declining to create a new snapshot at ${image.highestOffsetAndEpoch()} because " +
+ s"there is already a snapshot in progress at offset ${_currentSnapshotOffset}")
Review Comment:
Nit, the curly brackets ({}) are not needed for `_currentSnapshotOffset`. This comment applies to a few places.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -109,4 +115,9 @@ class BrokerMetadataSnapshotter(
beginShutdown()
eventQueue.close()
}
+
+ // VisibleForTesting
+ def currentSnapshotOffset(): Long = {
+ _currentSnapshotOffset
+ }
Review Comment:
It doesn't look like you use this in any of the tests. Let's remove this method if true.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,37 +45,32 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
- /**
- * The offset of the newest snapshot, or -1 if there hasn't been one. Accessed only under
- * the object lock.
- */
- private var _latestSnapshotOffset = -1L
-
/**
* The event queue which runs this listener.
*/
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
Review Comment:
So far we have two reasons for creating a snapshot. 1. X bytes were applied. 2. the metadata version changed. What do you think if we pass the reason for creating a snapshot to this method and logging it?
We would need to make a similar change to the KRaft Controller logic for creating a snapshot. If you agree, I am okay creating a Jira for this and implementing this in a future PR.
--
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] divijvaidya commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r912772869
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(100L, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV2.featureLevel(), 100L)
Review Comment:
> because this test will only be executed in a higher version Kafka
Why is that true? This change could potentially be back ported to 3.2.x version as well with the next patch at which stage it will use a different metadata version.
--
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 #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
jsancio commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r916060600
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,32 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (shouldSnapshot()) {
+ maybeStartSnapshot()
}
- snapshotter.foreach { snapshotter =>
- _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot() || metadataVersionChanged) {
- if (snapshotter.maybeStartSnapshot(_highestTimestamp, _delta.apply())) {
- _bytesSinceLastSnapshot = 0L
- }
- }
- }
+ _publisher.foreach(publish)
}
}
private def shouldSnapshot(): Boolean = {
- _bytesSinceLastSnapshot >= maxBytesBetweenSnapshots
+ (_bytesSinceLastSnapshot >= maxBytesBetweenSnapshots) || metadataVersionChanged()
+ }
+
+ private def metadataVersionChanged(): Boolean = {
+ _publisher.nonEmpty && Option(_delta.featuresDelta()).exists { featuresDelta =>
Review Comment:
I see. Before this PR the broker was generating a snapshot when handling committed records that didn't contain a feature change because we only reset `_delta` after the listener calls `publish`. This PR makes sense to me. Let's document this decision in this method.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r915532837
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,26 +45,37 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
+ /**
+ * The offset of the newest snapshot, or -1 if there hasn't been one. Accessed only under
+ * the object lock.
+ */
+ private var _latestSnapshotOffset = -1L
Review Comment:
This seems an existing problem, I checked the logic of `QuorumController.SnapshotGeneratorManager`, it will check the result of `raftClient.createSnapshot` and just skip it if it's empty, however, at broker side, `BrokerSnapshotWriterBuilder` will throw an exception on empty snapshot. how about making it consistent with controller code.
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r911771523
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+ assertEquals(100L, snapshotter.activeSnapshotOffset, "We should try to generate snapshot when starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChange(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+ listener.startPublishing(new MockMetadataPublisher()).get()
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV2.featureLevel(), 100L)
Review Comment:
I guess you are suggesting we should ensure the current metadataVersion is different from that we are updating to(IBP_3_3_IV2), this can be ensured because this test will only be executed in a higher version Kafka, and there isn't a reference to metadataVersion in `BrokerMetadataListener` so this is arduous 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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r916661591
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,37 +45,32 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
- /**
- * The offset of the newest snapshot, or -1 if there hasn't been one. Accessed only under
- * the object lock.
- */
- private var _latestSnapshotOffset = -1L
-
/**
* The event queue which runs this listener.
*/
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
Review Comment:
Good catch, I will create a ticket for this job.
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -51,20 +51,26 @@ class BrokerMetadataSnapshotter(
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
- if (_currentSnapshotOffset == -1L) {
+ if (_currentSnapshotOffset != -1) {
+ info(s"Declining to create a new snapshot at ${image.highestOffsetAndEpoch()} because " +
+ s"there is already a snapshot in progress at offset ${_currentSnapshotOffset}")
Review Comment:
This seems a bug of scala, if I remove the curly brackets there will be a compile error.
--
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] dengziming commented on pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on PR #12265:
URL: https://github.com/apache/kafka/pull/12265#issuecomment-1149647521
Hello, @mumrah the first bug is related to KIP-778, @jsancio you may be familiar with the second and third bugs, my solution for these problems are as follows:
1. Do not generate snapshots before publishing and try to generate a snapshot when starting publishing.
2. invoke `maybeStartSnapshot()` before `publish()`
--
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] dengziming commented on a diff in pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
dengziming commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r911778337
##########
core/src/test/scala/unit/kafka/server/metadata/BrokerMetadataListenerTest.scala:
##########
@@ -240,6 +239,40 @@ class BrokerMetadataListenerTest {
}
}
+ @Test
+ def testNotSnapshotAfterMetadataVersionChangeBeforePublishing(): Unit = {
+ val snapshotter = new MockMetadataSnapshotter()
+ val listener = newBrokerMetadataListener(snapshotter = Some(snapshotter),
+ maxBytesBetweenSnapshots = 1000L)
+
+ updateFeature(listener, feature = MetadataVersion.FEATURE_NAME, MetadataVersion.latest.featureLevel(), 100L)
+ listener.getImageRecords().get()
+ assertEquals(-1L, snapshotter.activeSnapshotOffset, "We won't generate snapshot before starting publishing")
+ }
+
+ @Test
+ def testSnapshotAfterMetadataVersionChangeWhenStarting(): Unit = {
Review Comment:
In a real cluster context, `handleSnaphot` and `handleCommit` are invoked from the Raft listener, and in this test case, they can only be invoked by us. since we only invoke `handleCommit` in this test I don't think it's necessary to check `handleSnapshot`.
--
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 #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r907665218
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##########
@@ -117,26 +117,30 @@ class BrokerMetadataListener(
} finally {
reader.close()
}
- _publisher.foreach(publish)
- // If we detected a change in metadata.version, generate a local snapshot
- val metadataVersionChanged = Option(_delta.featuresDelta()).exists { featuresDelta =>
- featuresDelta.metadataVersionChange().isPresent
+ _bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
+ if (_publisher.nonEmpty && shouldSnapshot()) {
Review Comment:
Perhaps we could move the publisher check into `metadataVersionChanged()`? My understanding is that `metadataVersionChanged` will always returns true until we have applied the first delta, so that would make the dependence on the publisher a little clearer. Do I have that right?
A consequence of changing that is that we could publish snapshots before the publisher has started if enough bytes have accumulated, which was the prior behavior. That would mean that we wouldn't need to update `StartPublishingEvent` below. Do you see any issues with that?
--
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 #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12265:
URL: https://github.com/apache/kafka/pull/12265#discussion_r915323373
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,26 +45,37 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
+ /**
+ * The offset of the newest snapshot, or -1 if there hasn't been one. Accessed only under
+ * the object lock.
+ */
+ private var _latestSnapshotOffset = -1L
Review Comment:
Do we need to initialize this on startup? If we don't, is it possible that we would end up snapshotting the same offset more than once?
##########
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##########
@@ -45,26 +45,37 @@ class BrokerMetadataSnapshotter(
*/
private var _currentSnapshotOffset = -1L
+ /**
+ * The offset of the newest snapshot, or -1 if there hasn't been one. Accessed only under
+ * the object lock.
+ */
+ private var _latestSnapshotOffset = -1L
+
/**
* The event queue which runs this listener.
*/
val eventQueue = new KafkaEventQueue(time, logContext, threadNamePrefix.getOrElse(""))
override def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage): Boolean = synchronized {
- if (_currentSnapshotOffset == -1L) {
+ if (_currentSnapshotOffset != -1) {
+ warn(s"Declining to create a new snapshot at ${image.highestOffsetAndEpoch()} because " +
Review Comment:
Do you think the log level is right for these messages? It seems like these are "normal" things that can happen and not an indication of a problem. Perhaps we can make them info or debug level.
--
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 pull request #12265: KAFKA-13968: Fix 3 major bugs of KRaft snapshot generating
Posted by GitBox <gi...@apache.org>.
jsancio commented on PR #12265:
URL: https://github.com/apache/kafka/pull/12265#issuecomment-1178979789
Reran the Jenkins build. It looks like the build is timing out. Maybe this is because we have a lot more tests after migrating some tests to run against KRaft and ZK.
--
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