You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mimaison (via GitHub)" <gi...@apache.org> on 2024/08/19 10:32:48 UTC
[PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
mimaison opened a new pull request, #16917:
URL: https://github.com/apache/kafka/pull/16917
### 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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1723026424
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -448,14 +449,16 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
/**
* Truncate the checkpointed offset for the given partition if its checkpointed offset is larger than the given offset
*/
- def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: Long): Unit = {
+ def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: JLong): Unit = {
inLock(lock) {
if (logs.get(topicPartition).config.compact) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
val existing = checkpoint.read()
- if (existing.getOrElse(topicPartition, 0L) > offset)
- checkpoint.write(mutable.Map() ++= existing += topicPartition -> offset)
+ if (existing.getOrDefault(topicPartition, 0L) > offset) {
+ existing.put(topicPartition, offset)
Review Comment:
That's a good point! I've added a comment
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1721581770
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -376,13 +377,13 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
* @param partitionToRemove The TopicPartition to be removed
*/
def updateCheckpoints(dataDir: File,
- partitionToUpdateOrAdd: Option[(TopicPartition, Long)] = None,
+ partitionToUpdateOrAdd: Option[(TopicPartition, JLong)] = None,
partitionToRemove: Option[TopicPartition] = None): Unit = {
inLock(lock) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
try {
- val currentCheckpoint = checkpoint.read().filter { case (tp, _) => logs.keys.contains(tp) }.toMap
+ val currentCheckpoint = checkpoint.read().asScala.filter { case (tp, _) => logs.keys.contains(tp) }.toMap
Review Comment:
It's unfortunate we have to convert the map into Scala here and back to Java below but keeping it in Java is really a pain.
Here is what I had to do to make it work:
```
val currentCheckpoint: util.Map[TopicPartition, JLong] = checkpoint.read().entrySet.stream
.filter(entry => logs.keys.contains(entry.getKey))
.collect(Collectors.toMap((entry: util.Map.Entry[TopicPartition, JLong]) => entry.getKey, (entry: util.Map.Entry[TopicPartition, JLong]) => entry.getValue))
// remove the partition offset if any
var updatedCheckpoint = partitionToRemove match {
case Some(topicPartition) =>
currentCheckpoint.remove(topicPartition)
currentCheckpoint
case None => currentCheckpoint
}
// update or add the partition offset if any
updatedCheckpoint = partitionToUpdateOrAdd match {
case Some(updatedOffset) =>
updatedCheckpoint.put(updatedOffset._1, updatedOffset._2)
updatedCheckpoint
case None => updatedCheckpoint
}
checkpoint.write(updatedCheckpoint)
```
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1721581770
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -376,13 +377,13 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
* @param partitionToRemove The TopicPartition to be removed
*/
def updateCheckpoints(dataDir: File,
- partitionToUpdateOrAdd: Option[(TopicPartition, Long)] = None,
+ partitionToUpdateOrAdd: Option[(TopicPartition, JLong)] = None,
partitionToRemove: Option[TopicPartition] = None): Unit = {
inLock(lock) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
try {
- val currentCheckpoint = checkpoint.read().filter { case (tp, _) => logs.keys.contains(tp) }.toMap
+ val currentCheckpoint = checkpoint.read().asScala.filter { case (tp, _) => logs.keys.contains(tp) }.toMap
Review Comment:
It's unfortunate we have to convert the map into Scala here and back to Java below but keeping it in Java is really a pain.
Here is what I had to do to make it work:
```
val currentCheckpoint: util.Map[TopicPartition, JLong] = checkpoint.read().entrySet.stream
.filter(entry => logs.keys.contains(entry.getKey))
.collect(Collectors.toMap(
(entry: util.Map.Entry[TopicPartition, JLong]) => entry.getKey,
(entry: util.Map.Entry[TopicPartition, JLong]) => entry.getValue))
// remove the partition offset if any
var updatedCheckpoint = partitionToRemove match {
case Some(topicPartition) =>
currentCheckpoint.remove(topicPartition)
currentCheckpoint
case None => currentCheckpoint
}
// update or add the partition offset if any
updatedCheckpoint = partitionToUpdateOrAdd match {
case Some(updatedOffset) =>
updatedCheckpoint.put(updatedOffset._1, updatedOffset._2)
updatedCheckpoint
case None => updatedCheckpoint
}
checkpoint.write(updatedCheckpoint)
```
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1721581770
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -376,13 +377,13 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
* @param partitionToRemove The TopicPartition to be removed
*/
def updateCheckpoints(dataDir: File,
- partitionToUpdateOrAdd: Option[(TopicPartition, Long)] = None,
+ partitionToUpdateOrAdd: Option[(TopicPartition, JLong)] = None,
partitionToRemove: Option[TopicPartition] = None): Unit = {
inLock(lock) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
try {
- val currentCheckpoint = checkpoint.read().filter { case (tp, _) => logs.keys.contains(tp) }.toMap
+ val currentCheckpoint = checkpoint.read().asScala.filter { case (tp, _) => logs.keys.contains(tp) }.toMap
Review Comment:
It's unfortunate we have to convert the map into Scala here and back to Java below but keeping it in Java is really a pain.
Here is what I had to do to make it work:
```scala
val currentCheckpoint: util.Map[TopicPartition, JLong] = checkpoint.read().entrySet.stream
.filter(entry => logs.keys.contains(entry.getKey))
.collect(Collectors.toMap(
(entry: util.Map.Entry[TopicPartition, JLong]) => entry.getKey,
(entry: util.Map.Entry[TopicPartition, JLong]) => entry.getValue))
// remove the partition offset if any
var updatedCheckpoint = partitionToRemove match {
case Some(topicPartition) =>
currentCheckpoint.remove(topicPartition)
currentCheckpoint
case None => currentCheckpoint
}
// update or add the partition offset if any
updatedCheckpoint = partitionToUpdateOrAdd match {
case Some(updatedOffset) =>
updatedCheckpoint.put(updatedOffset._1, updatedOffset._2)
updatedCheckpoint
case None => updatedCheckpoint
}
checkpoint.write(updatedCheckpoint)
```
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on PR #16917:
URL: https://github.com/apache/kafka/pull/16917#issuecomment-2299004441
None of the test failures are related, merging to trunk
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1722262658
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -448,14 +449,16 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
/**
* Truncate the checkpointed offset for the given partition if its checkpointed offset is larger than the given offset
*/
- def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: Long): Unit = {
+ def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: JLong): Unit = {
inLock(lock) {
if (logs.get(topicPartition).config.compact) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
Review Comment:
this is unrelated to this PR, but the scala map will throw `NoSuchElementException` when there is no mapping value from the given key. Hence, the null check is redundant.
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -448,14 +449,16 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
/**
* Truncate the checkpointed offset for the given partition if its checkpointed offset is larger than the given offset
*/
- def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: Long): Unit = {
+ def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: JLong): Unit = {
inLock(lock) {
if (logs.get(topicPartition).config.compact) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
val existing = checkpoint.read()
- if (existing.getOrElse(topicPartition, 0L) > offset)
- checkpoint.write(mutable.Map() ++= existing += topicPartition -> offset)
+ if (existing.getOrDefault(topicPartition, 0L) > offset) {
+ existing.put(topicPartition, offset)
Review Comment:
could you please add comments to `OffsetCheckpointFile#read` to remind the returned map should be mutable.
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1723024169
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -448,14 +449,16 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
/**
* Truncate the checkpointed offset for the given partition if its checkpointed offset is larger than the given offset
*/
- def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: Long): Unit = {
+ def maybeTruncateCheckpoint(dataDir: File, topicPartition: TopicPartition, offset: JLong): Unit = {
inLock(lock) {
if (logs.get(topicPartition).config.compact) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
Review Comment:
There's a couple of places with similar code. I'd prefer changing these in a separate 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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison merged PR #16917:
URL: https://github.com/apache/kafka/pull/16917
--
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
Re: [PR] MINOR: Move OffsetCheckpointFile to storage module [kafka]
Posted by "mimaison (via GitHub)" <gi...@apache.org>.
mimaison commented on code in PR #16917:
URL: https://github.com/apache/kafka/pull/16917#discussion_r1721581770
##########
core/src/main/scala/kafka/log/LogCleanerManager.scala:
##########
@@ -376,13 +377,13 @@ private[log] class LogCleanerManager(val logDirs: Seq[File],
* @param partitionToRemove The TopicPartition to be removed
*/
def updateCheckpoints(dataDir: File,
- partitionToUpdateOrAdd: Option[(TopicPartition, Long)] = None,
+ partitionToUpdateOrAdd: Option[(TopicPartition, JLong)] = None,
partitionToRemove: Option[TopicPartition] = None): Unit = {
inLock(lock) {
val checkpoint = checkpoints(dataDir)
if (checkpoint != null) {
try {
- val currentCheckpoint = checkpoint.read().filter { case (tp, _) => logs.keys.contains(tp) }.toMap
+ val currentCheckpoint = checkpoint.read().asScala.filter { case (tp, _) => logs.keys.contains(tp) }.toMap
Review Comment:
It's unfortunate we have to convert the map into Scala here and back to Java below but keeping it in Java is really a pain.
Here is what I had to do to make it work:
```scala
val currentCheckpoint: util.Map[TopicPartition, JLong] = checkpoint.read()
.entrySet.stream
.filter(entry => logs.keys.contains(entry.getKey))
.collect(Collectors.toMap(
(entry: util.Map.Entry[TopicPartition, JLong]) => entry.getKey,
(entry: util.Map.Entry[TopicPartition, JLong]) => entry.getValue))
// remove the partition offset if any
var updatedCheckpoint = partitionToRemove match {
case Some(topicPartition) =>
currentCheckpoint.remove(topicPartition)
currentCheckpoint
case None => currentCheckpoint
}
// update or add the partition offset if any
updatedCheckpoint = partitionToUpdateOrAdd match {
case Some(updatedOffset) =>
updatedCheckpoint.put(updatedOffset._1, updatedOffset._2)
updatedCheckpoint
case None => updatedCheckpoint
}
checkpoint.write(updatedCheckpoint)
```
--
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