junrao commented on a change in pull request #11345:
URL: https://github.com/apache/kafka/pull/11345#discussion_r753539711
##########
File path: core/src/main/scala/kafka/log/UnifiedLog.scala
##########
@@ -1505,21 +1505,35 @@ class UnifiedLog(@volatile var logStartOffset: Long,
/**
* Flush all local log segments
*/
- def flush(): Unit = flush(logEndOffset)
+ def flushUpToAndExcludingLogEndOffset(): Unit = flush(logEndOffset, false)
Review comment:
We could just have a single method flushToLogEndOffset(boolean
inclusive)?
##########
File path: core/src/main/scala/kafka/log/UnifiedLog.scala
##########
@@ -1505,21 +1505,35 @@ class UnifiedLog(@volatile var logStartOffset: Long,
/**
* Flush all local log segments
*/
- def flush(): Unit = flush(logEndOffset)
+ def flushUpToAndExcludingLogEndOffset(): Unit = flush(logEndOffset, false)
+
+ /**
+ * Flush all local log segments including possible empty active segment.
+ *
+ * We have to pass logEngOffset + 1 to the `localLog.flush(offset: Long):
Unit` function to flush empty
+ * active segments, which is important to make sure we don't lose the empty
index file during shutdown.
+ *
+ * This function should only be called during shutdown.
+ */
+ def flushUpToAndIncludingLogEndOffset(): Unit = flush(logEndOffset, true)
/**
* 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 flush(offset: Long): Unit = flush(offset, false)
+
+ private def flush(offset: Long, includingOffset: Boolean): Unit = {
+ val flushOffset = if (includingOffset) offset + 1 else offset
+ val recoveryPoint = offset
+ maybeHandleIOException(s"Error while flushing log for $topicPartition in
dir ${dir.getParent} with offset $flushOffset") {
+ if (flushOffset > localLog.recoveryPoint) {
+ debug(s"Flushing log up to offset $flushOffset, last flushed:
$lastFlushTime, current time: ${time.milliseconds()}, " +
Review comment:
Since recoveryPoint is not always flushOffset, could we log
recoveryPoint too?
##########
File path: core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
##########
@@ -212,7 +212,7 @@ final class KafkaMetadataLog private (
}
override def flush(): Unit = {
- log.flush()
+ log.flushUpToAndExcludingLogEndOffset()
Review comment:
It seems that the Raft log could have the same issue during a clean
shutdown. The easiest fix is to flush it inclusively during close.
--
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
