[GitHub] [kafka] jsancio commented on a diff in pull request #12414: KAFKA-14073 Logging the reason for Snapshot
jsancio commented on code in PR #12414:
URL: https://github.com/apache/kafka/pull/12414#discussion_r959867602
##
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##
@@ -119,16 +120,29 @@ class BrokerMetadataListener(
}
_bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot()) {
-maybeStartSnapshot()
+
+ val shouldTakeSnapshot: Set[SnapshotReason] = shouldSnapshot()
+ if (shouldTakeSnapshot.nonEmpty) {
+maybeStartSnapshot(shouldTakeSnapshot)
}
_publisher.foreach(publish)
}
}
- private def shouldSnapshot(): Boolean = {
-(_bytesSinceLastSnapshot >= maxBytesBetweenSnapshots) ||
metadataVersionChanged()
+ private def shouldSnapshot(): Set[SnapshotReason] = {
+val metadataVersionHasChanged: Boolean = metadataVersionChanged()
+val maxBytesHaveExceeded: Boolean = (_bytesSinceLastSnapshot >=
maxBytesBetweenSnapshots)
Review Comment:
Minor but we tend not to declare the type for private variables with simple
expressions.
##
core/src/main/scala/kafka/server/metadata/MetadataSnapshotter.scala:
##
@@ -31,5 +32,5 @@ trait MetadataSnapshotter {
*
* @return True if we will write out a new snapshot;
false otherwise.
*/
- def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage):
Boolean
+ def maybeStartSnapshot(lastContainedLogTime: Long, image: MetadataImage,
reason: Set[SnapshotReason]): Boolean
Review Comment:
Let's add the new parameter to the documentation above.
##
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##
@@ -59,11 +60,11 @@ class BrokerMetadataSnapshotter(
val writer = writerBuilder.build(
image.highestOffsetAndEpoch().offset,
image.highestOffsetAndEpoch().epoch,
-lastContainedLogTime
- )
+lastContainedLogTime)
Review Comment:
Minor but you can revert this change. It is okay to have the closing
delimiter in a new line.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12414: KAFKA-14073 Logging the reason for Snapshot
jsancio commented on code in PR #12414:
URL: https://github.com/apache/kafka/pull/12414#discussion_r937029349
##
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##
@@ -129,16 +129,17 @@ class BrokerMetadataListener(
}
}
- private def shouldSnapshot(): Option[String] = {
+ private def shouldSnapshot(): Option[SnapshotReason] = {
val metadataVersionHasChanged: Boolean = metadataVersionChanged()
val maxBytesHaveExceeded: Boolean = (_bytesSinceLastSnapshot >=
maxBytesBetweenSnapshots)
+//TODO: Figure out how to handle multiple reasons for snapshot
Review Comment:
One solution is to return a `Set[SnapshotReason]` and pass the set when
creating the snapshot.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] jsancio commented on a diff in pull request #12414: KAFKA-14073 Logging the reason for Snapshot
jsancio commented on code in PR #12414:
URL: https://github.com/apache/kafka/pull/12414#discussion_r934580125
##
core/src/main/scala/kafka/server/metadata/BrokerMetadataSnapshotter.scala:
##
@@ -63,7 +63,7 @@ class BrokerMetadataSnapshotter(
)
if (writer.nonEmpty) {
_currentSnapshotOffset = image.highestOffsetAndEpoch().offset
-info(s"Creating a new snapshot at offset ${_currentSnapshotOffset}...")
+info(s"Creating a new snapshot at offset ${_currentSnapshotOffset}
because ${snapshotReason}...")
Review Comment:
What do you think about moving this log line to the `KafkaMetadataLog` type?
That code is shared between the Controller and the Broker.
##
core/src/main/scala/kafka/server/metadata/BrokerMetadataListener.scala:
##
@@ -119,16 +119,29 @@ class BrokerMetadataListener(
}
_bytesSinceLastSnapshot = _bytesSinceLastSnapshot + results.numBytes
- if (shouldSnapshot()) {
-maybeStartSnapshot()
+
+ val shouldTakeSnapshot: Option[String] = shouldSnapshot()
+ if (shouldTakeSnapshot.isDefined) {
+maybeStartSnapshot(shouldTakeSnapshot.get)
}
_publisher.foreach(publish)
}
}
- private def shouldSnapshot(): Boolean = {
-(_bytesSinceLastSnapshot >= maxBytesBetweenSnapshots) ||
metadataVersionChanged()
+ private def shouldSnapshot(): Option[String] = {
Review Comment:
Let's have a stricter type. I suspect that we are going to add more reasons
for generating snapshots. What do you think if we create a Java enum
`SnapshotReason` in the `raft` module and `org.apache.kafka.snapshot` package?
We can also changed the signature of `RaftClient#creaeSnapshot` to
```
Optional> createSnapshot(
long committedOffset,
int committedEpoch,
long lastContainedLogTime,
SnapshotReason reason
);
```
You can take a look at `LogStartOffsetIncrementReason` in
`core/src/main/scala/kafka/log/UnifiedLog.scala` for an example in Scala. We
want this as an enum in Java since it will be used in `KafkaRaftClient`.
##
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##
@@ -1325,7 +1325,7 @@ private void maybeGenerateSnapshot(long batchSizeInBytes)
{
snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
}
-log.info("Generating a snapshot that includes (epoch={},
offset={}) after {} committed bytes since the last snapshot.",
+log.info("Generating a snapshot that includes (epoch={},
offset={}) after {} committed bytes since the last snapshot, because max bytes
exceeded.",
Review Comment:
What do you think about moving this log line to the `KafkaMetadataLog` type?
That code is shared between the Controller and the Broker.
##
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##
@@ -1325,7 +1325,7 @@ private void maybeGenerateSnapshot(long batchSizeInBytes)
{
snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
}
-log.info("Generating a snapshot that includes (epoch={},
offset={}) after {} committed bytes since the last snapshot.",
+log.info("Generating a snapshot that includes (epoch={},
offset={}) after {} committed bytes since the last snapshot, because max bytes
exceeded.",
Review Comment:
Hmm. So it looks like in the Controller we don't generate a snapshot when
the metadata version changes.
@mumrah How should we resolve 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
