KAFKA-7366: Make topic configs segment.bytes and segment.ms to take effect immediately #5728
Conversation
|
@junrao Please take a look, whenever you get a chance |
|
|
||
| (1 to 50).foreach(i => TestUtils.produceMessage(servers, tp.topic, i.toString)) | ||
| // Verify that the new config is used for all segments | ||
| TestUtils.waitUntilTrue(() => log.logSegments.forall(_.size > 1000), "Log segment size change not applied") |
There was a problem hiding this comment.
Do we need the waitUntil part? The default producer acks is 1. So, by the time that TestUtils.produceMessage returns, the data should be in the log segments.
There was a problem hiding this comment.
@junrao Thanks for the review. Updated the PR.
| @@ -181,21 +180,23 @@ class LogSegmentTest { | |||
| assertFalse(reopened.timeIndex.isFull) | |||
| assertFalse(reopened.offsetIndex.isFull) | |||
|
|
|||
| assertFalse(reopened.shouldRoll(messagesSize = 1024, | |||
| val logConfig = LogConfig(Map().asJava) | |||
There was a problem hiding this comment.
Shouldn't we be setting the relevant configs here? Previously we were passing maxSegmentMs to LogSegment and setting maxSegmentBytes to Int.MaxValue.
There was a problem hiding this comment.
Thanks for catching this. we should pass the configs here.
| def shouldRoll(messagesSize: Int, maxTimestampInMessages: Long, maxOffsetInMessages: Long, now: Long): Boolean = { | ||
| val reachedRollMs = timeWaitedForRoll(now, maxTimestampInMessages) > maxSegmentMs - rollJitterMs | ||
| size > maxSegmentBytes - messagesSize || | ||
| def shouldRoll(config: LogConfig, messagesSize: Int, maxTimestampInMessages: Long, maxOffsetInMessages: Long, now: Long): Boolean = { |
There was a problem hiding this comment.
Hmm, why are we passing the whole config when we only need two parameters? One thing we should consider is introducing a RollParams case class to make this more maintainable. Generally, it makes it harder to test and understand if we pass these config classes with lots of params when we only need a subset.
There was a problem hiding this comment.
RollParams could have an apply that takes AppendInfo and LogConfig which could be used by Log.maybeRoll.
There was a problem hiding this comment.
@ijuma Thanks for the review. Yes, we can add a RollParams class to hold the params.
Added a commit which implements suggested approach. Pls take take a look, when you get a chance.
omkreddy
force-pushed
the
KAFKA-7366-Segment-bytes
branch
from
8a5e64a to
0729657
Compare
| LogConfig.SegmentBytesProp -> Int.MaxValue | ||
| ).asJava) | ||
|
|
||
| var rollParams = RollParams(logConfig, getLogAppendInfo(100L), 1024, time.milliseconds()) |
There was a problem hiding this comment.
Why don't we use the RollParams that takes the values explicitly instead of creating LogConfig and AppendInfo in this case? It doesn't seem like it helps in this case.
omkreddy
force-pushed
the
KAFKA-7366-Segment-bytes
branch
2 times, most recently
from
75a8922 to
82c26a2
Compare
omkreddy
force-pushed
the
KAFKA-7366-Segment-bytes
branch
from
82c26a2 to
b9643f4
Compare
…ffect immediately (#5728) Reviewers: Ismael Juma <[email protected]> and Jun Rao <[email protected]>
…ffect immediately (apache#5728) Reviewers: Ismael Juma <[email protected]> and Jun Rao <[email protected]>
Committer Checklist (excluded from commit message)