junrao commented on a change in pull request #11345:
URL: https://github.com/apache/kafka/pull/11345#discussion_r781653267
##########
File path: core/src/main/scala/kafka/log/UnifiedLog.scala
##########
@@ -1498,28 +1498,46 @@ class UnifiedLog(@volatile var logStartOffset: Long,
producerStateManager.takeSnapshot()
updateHighWatermarkWithLogEndOffset()
// Schedule an asynchronous flush of the old segment
- scheduler.schedule("flush-log", () => flush(newSegment.baseOffset))
+ scheduler.schedule("flush-log", () =>
flushUptoOffsetExclusive(newSegment.baseOffset))
newSegment
}
/**
* Flush all local log segments
+ *
+ * @param forceFlushActiveSegment should be true during a clean shutdown,
and false otherwise. The reason is that
+ * we have to pass logEndOffset + 1 to the `localLog.flush(offset: Long):
Unit` function to flush empty
+ * active segments, which is important to make sure we persist the active
segment file during shutdown, particularly
+ * when it's empty.
*/
- def flush(): Unit = flush(logEndOffset)
+ def flush(forceFlushActiveSegment: Boolean): Unit = flush(logEndOffset,
forceFlushActiveSegment)
/**
* Flush local log segments for all offsets up to offset-1
*
* @param offset The offset to flush up to (non-inclusive); the new recovery
point
*/
- def flush(offset: Long): Unit = {
- maybeHandleIOException(s"Error while flushing log for $topicPartition in
dir ${dir.getParent} with offset $offset") {
- if (offset > localLog.recoveryPoint) {
- debug(s"Flushing log up to offset $offset, last flushed:
$lastFlushTime, current time: ${time.milliseconds()}, " +
+ def flushUptoOffsetExclusive(offset: Long): Unit = flush(offset, false)
+
+ /**
+ * Flush local log segments for all offsets up to offset-1 if
includingOffset=false; up to offset
+ * if includingOffset=true. The recovery point is set to offset.
+ *
+ * @param offset The offset to flush up to; the new recovery point
+ * @param includingOffset Whether the flush includes the provided offset.
+ */
+ private def flush(offset: Long, includingOffset: Boolean): Unit = {
+ val flushOffset = if (includingOffset) offset + 1 else offset
+ val newRecoveryPoint = offset
+ maybeHandleIOException(s"Error while flushing log for $topicPartition in
dir ${dir.getParent} with offset $offset ("
+ + { if (includingOffset) "inclusive" else "exclusive" }
+ + s") and recovery point $newRecoveryPoint") {
+ if (flushOffset > localLog.recoveryPoint) {
+ debug(s"Flushing log up to offset $flushOffset with recovery point
$newRecoveryPoint, last flushed: $lastFlushTime, current time:
${time.milliseconds()}, " +
Review comment:
Instead of $flushOffset, could we change to "$offset(ex/inclusive)"?
##########
File path: raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java
##########
@@ -184,8 +184,10 @@ default ValidOffsetAndEpoch validateOffsetAndEpoch(long
offset, int epoch) {
/**
* Flush the current log to disk.
+ *
+ * @param forceFlushActiveSegment Whether the flush includes the log end
offset. Should be `true` during close; otherwise false.
Review comment:
Whether the flush includes the log end offset => Whether to force flush
the active segment?
--
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
