KAFKA-20012: Move NodeToControllerChannelManagerImpl to server module #21261
Conversation
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@joshua2519 thanks for this patch!
| /** | ||
| * Discovers the active controller node and provides connection details for communicating with it. | ||
| */ | ||
| public interface ControllerNodeProvider { |
There was a problem hiding this comment.
Could we use Supplier<ControllerInformation> instead?
There was a problem hiding this comment.
Done. ControllerNodeProvider now extends Supplier for better idiomatic Java usage.
| * | ||
| * @param reconfigurable the component to register for configuration updates | ||
| */ | ||
| public void addReconfigurable(Reconfigurable reconfigurable) { |
There was a problem hiding this comment.
why we don't set it as abstract method?
There was a problem hiding this comment.
Agreed and changed. Since all current implementations (KafkaConfig) already provide concrete implementations, making them abstract clarifies the contract and ensures future subclasses don't accidentally rely on no-op behavior.
| * | ||
| * @param reconfigurable the component to unregister | ||
| */ | ||
| public void removeReconfigurable(Reconfigurable reconfigurable) { |
| private final ApiVersions apiVersions = new ApiVersions(); | ||
| private final NodeToControllerRequestThread requestThread; | ||
|
|
||
| @SuppressWarnings("this-escape") |
There was a problem hiding this comment.
Would you mind inlining newRequestThread to get rid of this warning?
| } else if (response.wasDisconnected()) { | ||
| updateControllerAddress(null); | ||
| try { | ||
| requestQueue.putFirst(queueItem); |
There was a problem hiding this comment.
We can substitute addFirst for putFirst, since requestQueue is an unbounded LinkedBlockingDeque
There was a problem hiding this comment.
Agreed. Since requestQueue is unbounded (LinkedBlockingDeque with no capacity limit), putFirst() will never block and the InterruptedException handling is unnecessary. Changed to addFirst() for simplicity.
| /** | ||
| * Discovers the active controller node and provides connection details for communicating with it. | ||
| */ | ||
| public interface ControllerNodeProvider extends Supplier<ControllerInformation> { |
There was a problem hiding this comment.
I meant that the standard Supplier<ControllerInformation> interface could simply replace ControllerNodeProvider
There was a problem hiding this comment.
Done. Replaced ControllerNodeProvider with Supplier throughout the codebase since there's only one production implementation.
| } | ||
|
|
||
| @Override | ||
| public void shutdown() { |
There was a problem hiding this comment.
We should throw an exception here to align with the Scala implementation
There was a problem hiding this comment.
Done. Updated shutdown() to declare throws InterruptedException allowing the exception to propagate instead of being swallowed.
| Optional<Node> controllerAddress = activeControllerAddress(); | ||
| if (controllerAddress.isPresent()) { | ||
| requestIter.remove(); | ||
| return Collections.singletonList(new RequestAndCompletionHandler( |
| } | ||
| } | ||
|
|
||
| return Collections.emptyList(); |
| activeControllerAddress().map(Node::idString).orElse("null")); | ||
| maybeDisconnectAndUpdateController(); | ||
| try { | ||
| requestQueue.putFirst(queueItem); |
| } | ||
|
|
||
| @SuppressWarnings("resource") // RaftClient lifecycle managed by RaftManager | ||
| private Optional<Node> idToNode(int id) { |
There was a problem hiding this comment.
Do we really need this private method? Having two @SuppressWarnings seems to double the noise
There was a problem hiding this comment.
Agreed, I have inlined to get().
3 participants
Migrates
NodeToControllerChannelManagerand related classes from Scala(core module) to Java (server module).
New Java Classes (server module) include
ControllerInformation,RaftControllerNodeProvider,NodeToControllerChannelManagerImpl,NodeToControllerRequestThreadandNodeToControllerQueueItem.In
AbstractKafkaConfig, addedaddReconfigurable()/removeReconfigurable()abstract methods tosupport migration. In
KafkaConfig, addedoverridekeywords forreconfigurable methods. Updated imports and usages across core module
classes and tests. Replace
ControllerNodeProviderbySupplier<ControllerInformation>.Reviewers: Chia-Ping Tsai [email protected]