feat(inkless): Init diskless log on Control Plane#563
Merged
Conversation
f505e6a to
d0e3bd5
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
This PR completes the classic-to-diskless init flow by triggering a control-plane initDisklessLog call once the committed metadata (diskless start offset + producer states) has been applied, finalizing the AwaitingMetadata portion of the orchestration.
Changes:
- Wire
ControlPlaneintoInitDisklessLogManagerand add anAwaitingMetadatabatch queue/protocol to initialize diskless logs on the control plane after metadata is committed. - Invoke the new control-plane init path from
ReplicaManager.applyDeltawhen diskless fields appear in aPartitionRegistration. - Add/extend unit and cluster-level tests covering the sealing → controller init → metadata commit → control-plane init end-to-end path.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| core/src/main/scala/kafka/server/BrokerServer.scala | Passes ControlPlane into InitDisklessLogManager during broker startup. |
| core/src/main/scala/kafka/server/InitDisklessLogBatchQueue.scala | Adds an AwaitingMetadataBatchQueue for control-plane calls. |
| core/src/main/scala/kafka/server/InitDisklessLogManager.scala | Adds initOnControlPlane and enqueues AwaitingMetadata states for control-plane init. |
| core/src/main/scala/kafka/server/InitDisklessLogState.scala | Extends AwaitingMetadata with payload + adds Done and a control-plane batch protocol. |
| core/src/main/scala/kafka/server/ReplicaManager.scala | Detects committed diskless fields in metadata deltas and triggers control-plane init. |
| core/src/test/scala/unit/kafka/server/InitDisklessLogManagerTest.scala | Adds unit tests for control-plane init success/error/idempotency behavior. |
| core/src/test/scala/unit/kafka/server/metadata/InitDisklessLogFlowTest.scala | Adds metadata-flow tests verifying control-plane init is triggered after PCR diskless fields commit. |
| core/src/test/java/kafka/server/InklessTopicTypeSwitcherClusterTest.java | Adds a Testcontainers-based cluster test for classic → diskless config migration while producing/consuming. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
7824c42 to
630cdab
Compare
Base automatically changed from
giuseppelillo/refactor-init-diskless-log-manager
to
main
April 8, 2026 14:59
7064e97 to
447e9ed
Compare
Complete the InitDisklessLogManager orchestration with the final transition from AwaitingMetadata to Done.
447e9ed to
7b16a9b
Compare
giuseppelillo
commented
Apr 9, 2026
Comment on lines
-240
to
-255
| if (maxAttemptNumber <= 0) { | ||
| retryPeriodMs | ||
| } else { | ||
| // First retry waits retryPeriodMs, then doubles exponentially up to maxRetryTimeMs. | ||
| var delay = retryPeriodMs | ||
| var exponent = maxAttemptNumber - 1 | ||
| while (exponent > 0 && delay < maxRetryTimeMs) { | ||
| if (delay >= (maxRetryTimeMs + 1L) / 2L) { | ||
| delay = maxRetryTimeMs | ||
| } else { | ||
| delay = delay * 2L | ||
| } | ||
| exponent -= 1 | ||
| } | ||
| Math.min(delay, maxRetryTimeMs) | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Refactored to use ExponentialBackoff instead.
giuseppelillo
commented
Apr 9, 2026
Comment on lines
-100
to
-115
| val preservedAttemptNumber = Option(queuedByTp.get(tp)).map(_.attemptNumber).getOrElse(0) | ||
| // Keep retry progression for already queued partitions, but always refresh the state payload. | ||
| queuedByTp.put(tp, Attempt(state, attemptNumber = preservedAttemptNumber)) | ||
| val currentPromise = resultPromiseByTp.computeIfAbsent(tp, _ => Promise[Boolean]()) | ||
|
|
||
| taskStatus match { | ||
| case NoTask => | ||
| // Schedule a new run, allowing linger for additional ready partitions. | ||
| scheduleNewTask(lingerMs) | ||
| case TaskScheduled(scheduledTask) => | ||
| // Re-schedule using linger so newly ready partitions can batch together. | ||
| scheduledTask.cancel(false) | ||
| scheduleNewTask(lingerMs) | ||
| case TaskRunning => | ||
| // A task is in-flight. Newly queued work is already in queuedByTp | ||
| // and will be picked up by finishTask. |
Contributor
Author
There was a problem hiding this comment.
Little refactor to allow to handle duplicate requests not yet sent.
jeqo
reviewed
Apr 9, 2026
jeqo
left a comment
Contributor
There was a problem hiding this comment.
Looks good to me, only a few minor improvements suggested.
jeqo
previously approved these changes
Apr 10, 2026
| Option(resultPromiseByTp.remove(tp)).foreach(_.trySuccess(false)) | ||
| def remove(tp: TopicPartition): Unit = { | ||
| withQueueLock { queuedByTp.remove(tp) } | ||
| completeAndRemovePromise(tp, accepted = false) |
Contributor
There was a problem hiding this comment.
why do you move it to outside the queueLock? It seems like resultPromiseByTp in completeAndRemovePromise is used without the lock, while in enqueue() it's in the lock. That way a promise can be added legitimately but another thread may remove it before it was supposed to be removed. Is this intended?
Contributor
Author
There was a problem hiding this comment.
You're right, this was the result of a refactor using a ConcurrentHashMap, but the race condition is still there. I've changed it now so the removal is done within the lock, while the promise completion can be done outside.
jeqo
approved these changes
Apr 10, 2026
jeqo
left a comment
Contributor
There was a problem hiding this comment.
LGTM. deferring to @viktorsomogyi for a final look and merge
viktorsomogyi
approved these changes
Apr 14, 2026
jeqo
pushed a commit
that referenced
this pull request
Apr 14, 2026
Complete the InitDisklessLogManager orchestration with the final transition from AwaitingMetadata to Done.
jeqo
pushed a commit
that referenced
this pull request
Apr 14, 2026
Complete the InitDisklessLogManager orchestration with the final transition from AwaitingMetadata to Done.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Complete the InitDisklessLogManager orchestration with the final transition from
AwaitingMetadatatoDone.