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/03/25 14:25:39 UTC
[GitHub] [kafka] yyu1993 opened a new pull request #11950: Fix missing active segment
yyu1993 opened a new pull request #11950:
URL: https://github.com/apache/kafka/pull/11950
In the log layer, if all segments from a log are deleted before a new segment is created, the log will have no active segment for a brief time. This will allows for a race condition to happen if other threads are reading the active segment at the same time which will can result in a NoSuchElementException.
This PR:
- introduces a new function which creates a new segment and then deletes the given segment. This would ensure the log will have an active segment at all times.
- fixes two places where segments are deleted before new segment is created.
### 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] junrao commented on a change in pull request #11950: KAFKA-12875: Change Log layer segment map mutations to avoid absence of active segment
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#discussion_r836679617
##########
File path: core/src/main/scala/kafka/log/LocalLog.scala
##########
@@ -517,14 +558,16 @@ class LocalLog(@volatile private var _dir: File,
debug(s"Truncate and start at offset $newOffset")
checkIfMemoryMappedBufferClosed()
val segmentsToDelete = List[LogSegment]() ++ segments.values
- removeAndDeleteSegments(segmentsToDelete, asyncDelete = true, LogTruncation(this))
- segments.add(LogSegment.open(dir,
- baseOffset = newOffset,
- config = config,
- time = time,
- initFileSize = config.initFileSize,
- preallocate = config.preallocate))
+ if (segmentsToDelete.size > 1)
Review comment:
Would it be better to combine the two ifs together to sth like the following? It would also be useful to add a comment why we want do special treatment for the last segment.
```
if (segmentsToDelete.nonEmpty) {
removeAndDeleteSegments(segmentsToDelete.dropRight(1), asyncDelete = true, LogTruncation(this))
createAndDeleteSegment(newOffset, segmentsToDelete.last, asyncDelete = true, LogTruncation(this))
}
```
##########
File path: core/src/main/scala/kafka/log/UnifiedLog.scala
##########
@@ -1355,15 +1355,19 @@ class UnifiedLog(@volatile var logStartOffset: Long,
val numToDelete = deletable.size
if (numToDelete > 0) {
// we must always have at least one segment, so if we are going to delete all the segments, create a new one first
- if (localLog.segments.numberOfSegments == numToDelete)
- roll()
- lock synchronized {
- localLog.checkIfMemoryMappedBufferClosed()
- // remove the segments for lookups
- localLog.removeAndDeleteSegments(deletable, asyncDelete = true, reason)
- deleteProducerSnapshots(deletable, asyncDelete = true)
- maybeIncrementLogStartOffset(localLog.segments.firstSegmentBaseOffset.get, SegmentDeletion)
+ var segmentsToDelete = deletable
+ if (localLog.segments.numberOfSegments == numToDelete) {
+ val newSegment = roll()
+ if (deletable.last.baseOffset == newSegment.baseOffset) {
+ warn(s"Empty active segment at ${deletable.last.baseOffset} was deleted and recreated due to $reason")
+ segmentsToDelete = deletable.dropRight(1)
Review comment:
Thanks, Yang. Makes sense to me now.
--
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] junrao merged pull request #11950: KAFKA-12875: Change Log layer segment map mutations to avoid absence of active segment
Posted by GitBox <gi...@apache.org>.
junrao merged pull request #11950:
URL: https://github.com/apache/kafka/pull/11950
--
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] yyu1993 commented on pull request #11950: Fix missing active segment
Posted by GitBox <gi...@apache.org>.
yyu1993 commented on pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#issuecomment-1079087291
Hi @kowshik and @junrao
could you help review this PR? This is same as the fix we made for ce-kafka.
--
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] yyu1993 edited a comment on pull request #11950: Fix missing active segment
Posted by GitBox <gi...@apache.org>.
yyu1993 edited a comment on pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#issuecomment-1079087291
Hi @kowshik and @junrao
could you help review this 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] kowshik commented on pull request #11950: Fix missing active segment
Posted by GitBox <gi...@apache.org>.
kowshik commented on pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#issuecomment-1079354780
@yyu1993 KAFKA-12875 is the jira we had created for this earlier, we could associate this PR with the same.
--
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] yyu1993 commented on a change in pull request #11950: KAFKA-12875: Change Log layer segment map mutations to avoid absence of active segment
Posted by GitBox <gi...@apache.org>.
yyu1993 commented on a change in pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#discussion_r835676088
##########
File path: core/src/main/scala/kafka/log/UnifiedLog.scala
##########
@@ -1355,15 +1355,19 @@ class UnifiedLog(@volatile var logStartOffset: Long,
val numToDelete = deletable.size
if (numToDelete > 0) {
// we must always have at least one segment, so if we are going to delete all the segments, create a new one first
- if (localLog.segments.numberOfSegments == numToDelete)
- roll()
- lock synchronized {
- localLog.checkIfMemoryMappedBufferClosed()
- // remove the segments for lookups
- localLog.removeAndDeleteSegments(deletable, asyncDelete = true, reason)
- deleteProducerSnapshots(deletable, asyncDelete = true)
- maybeIncrementLogStartOffset(localLog.segments.firstSegmentBaseOffset.get, SegmentDeletion)
+ var segmentsToDelete = deletable
+ if (localLog.segments.numberOfSegments == numToDelete) {
+ val newSegment = roll()
+ if (deletable.last.baseOffset == newSegment.baseOffset) {
+ warn(s"Empty active segment at ${deletable.last.baseOffset} was deleted and recreated due to $reason")
+ segmentsToDelete = deletable.dropRight(1)
Review comment:
In roll(), we only call createAndDeleteSegment() when new segment's offset is same as the last segment's offset and last segment is empty. So in the case when new segment's offset is different, we will not delete the last segment in roll(). It will be deleted as part of segmentsToDelete.
--
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] yyu1993 commented on pull request #11950: KAFKA-12875: Change Log layer segment map mutations to avoid absence of active segment
Posted by GitBox <gi...@apache.org>.
yyu1993 commented on pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#issuecomment-1084831187
> @yyu1993 : Thanks for the explanation. Just one more comment. Also, are the test failures related?
Hi @junrao
The tests passed in the rerun.
One additional test failed: kafka.admin.TopicCommandIntegrationTest.testDescribeUnderMinIsrPartitionsMixed() for JDK 8 and Scala 2.12, but this test passed on the previous run and does not seem to relate to this 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] junrao commented on a change in pull request #11950: Fix missing active segment
Posted by GitBox <gi...@apache.org>.
junrao commented on a change in pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#discussion_r835434811
##########
File path: core/src/main/scala/kafka/log/UnifiedLog.scala
##########
@@ -1355,15 +1355,19 @@ class UnifiedLog(@volatile var logStartOffset: Long,
val numToDelete = deletable.size
if (numToDelete > 0) {
// we must always have at least one segment, so if we are going to delete all the segments, create a new one first
- if (localLog.segments.numberOfSegments == numToDelete)
- roll()
- lock synchronized {
- localLog.checkIfMemoryMappedBufferClosed()
- // remove the segments for lookups
- localLog.removeAndDeleteSegments(deletable, asyncDelete = true, reason)
- deleteProducerSnapshots(deletable, asyncDelete = true)
- maybeIncrementLogStartOffset(localLog.segments.firstSegmentBaseOffset.get, SegmentDeletion)
+ var segmentsToDelete = deletable
+ if (localLog.segments.numberOfSegments == numToDelete) {
+ val newSegment = roll()
+ if (deletable.last.baseOffset == newSegment.baseOffset) {
+ warn(s"Empty active segment at ${deletable.last.baseOffset} was deleted and recreated due to $reason")
+ segmentsToDelete = deletable.dropRight(1)
Review comment:
Hmm, I am not sure if this part of the logic is completely right. roll() eventually calls createAndDeleteSegment() that always deletes the current active segment. So, in the case when the new segment's offset is different from the last segment's offset, it seems we will be deleting the last segment twice, once in createAndDeleteSegment() and another 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] yyu1993 commented on a change in pull request #11950: KAFKA-12875: Change Log layer segment map mutations to avoid absence of active segment
Posted by GitBox <gi...@apache.org>.
yyu1993 commented on a change in pull request #11950:
URL: https://github.com/apache/kafka/pull/11950#discussion_r837913065
##########
File path: core/src/main/scala/kafka/log/LocalLog.scala
##########
@@ -517,14 +558,16 @@ class LocalLog(@volatile private var _dir: File,
debug(s"Truncate and start at offset $newOffset")
checkIfMemoryMappedBufferClosed()
val segmentsToDelete = List[LogSegment]() ++ segments.values
- removeAndDeleteSegments(segmentsToDelete, asyncDelete = true, LogTruncation(this))
- segments.add(LogSegment.open(dir,
- baseOffset = newOffset,
- config = config,
- time = time,
- initFileSize = config.initFileSize,
- preallocate = config.preallocate))
+ if (segmentsToDelete.size > 1)
Review comment:
Makes sense. removeAndDeleteSegments has a check for empty lists, so it makes sense to combine the if statements.
--
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