KAFKA-15369: Implement KIP-919: Allow AC to Talk Directly with Controllers by cmccabe · Pull Request #14306 · apache/kafka

cmccabe · 2023-08-29T05:54:39Z

Implement KIP-919: Allow AdminClient to Talk Directly with the KRaft Controller Quorum and add Controller Registration. This KIP adds a new version of DescribeClusterRequest which is supported by KRaft controllers. It also teaches AdminClient how to use this new DESCRIBE_CLUSTER request to talk directly with the controller quorum. This is all gated behind a new MetadataVersion, IBP_3_7_IV0.

In order to share the DESCRIBE_CLUSTER logic between broker and controller, this PR factors it out into AuthHelper.computeDescribeClusterResponse.

The KIP adds three new errors codes: MISMATCHED_ENDPOINT_TYPE, UNSUPPORTED_ENDPOINT_TYPE, and UNKNOWN_CONTROLLER_ID. The endpoint type errors can be returned from DescribeClusterRequest

On the controller side, the controllers now try to register themselves with the current active controller, by sending a CONTROLLER_REGISTRATION request. This, in turn, is converted into a RegisterControllerRecord by the active controller. ClusterImage, ClusterDelta, and all other associated classes have been upgraded to propagate the new metadata. In the metadata shell, the cluster directory now contains both broker and controller subdirectories.

QuorumFeatures previously had a reference to the ApiVersions structure used by the controller's NetworkClient. Because this PR removes that reference, QuorumFeatures now contains only immutable data. Specifically, it contains the current node ID, the locally supported features, and the list of quorum node IDs in the cluster.

…llers Implement KIP-919: Allow AdminClient to Talk Directly with the KRaft Controller Quorum and add Controller Registration. This KIP adds a new version of DescribeClusterRequest which is supported by KRaft controllers. It also teaches AdminClient how to use this new DESCRIBE_CLUSTER request to talk directly with the controller quorum. This is all gated behind a new MetadataVersion, IBP_3_7_IV0. In order to share the DESCRIBE_CLUSTER logic between broker and controller, this PR factors it out into AuthHelper.computeDescribeClusterResponse. The KIP adds three new errors codes: MISMATCHED_ENDPOINT_TYPE, UNSUPPORTED_ENDPOINT_TYPE, and UNKNOWN_CONTROLLER_ID. The endpoint type errors can be returned from DescribeClusterRequest On the controller side, the controllers now try to register themselves with the current active controller, by sending a CONTROLLER_REGISTRATION request. This, in turn, is converted into a RegisterControllerRecord by the active controller. ClusterImage, ClusterDelta, and all other associated classes have been upgraded to propagate the new metadata. In the metadata shell, the cluster directory now contains both broker and controller subdirectories. QuorumFeatures previously had a reference to the ApiVersions structure used by the controller's NetworkClient. Because this PR removes that reference, QuorumFeatures now contains only immutable data. Specifically, it contains the current node ID, the locally supported features, and the list of quorum node IDs in the cluster.

showuon · 2023-08-31T10:02:07Z

-log4j.rootLogger=OFF, stdout
+log4j.rootLogger=DEBUG, stdout

 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n

-log4j.logger.kafka=WARN
-log4j.logger.org.apache.kafka=WARN
+log4j.logger.kafka=DEBUG
+log4j.logger.org.apache.kafka=DEBUG


Should we revert them?

yes, this should be reverted. I'll fix it now.

showuon · 2023-08-31T10:08:02Z

+  }
+
+  private def rpcStats(manager: ControllerRegistrationManager): (Long, Long, Long) = {
+    val failedAttempts = new CompletableFuture[(Long, Long, Long)]


What does the variable name failedAttempted mean here?

number of times we've tried to send an RPC and failed for some reason (or gotten back an error code)

mumrah

Thanks @cmccabe! Left some comments inline.

One question I had was about de-registration (or lack thereof). How do we deal with one controller being offline? Do we just use its most recent registration?

If two controllers are down, then no metadata gets updated (or published?) since we don't have a leader. But if we have two out of three, what happens if we make a decision based on the old registration and then that controller comes back up in a different configuration?

E.g.,

Three controllers A, B, C register at metadata.version X
Controller A goes down
Operator bumps up metadata.version to X+1, this works because the active controller looks at A's previous registration
Controller A comes back up with older software that doesn't support X (or maybe B and C upgrade to newer software)
Controller A replays FeatureLevelRecord X+1 and crashes

I vaguely recall discussing something like this during the KIP, but I just wanted to confirm the expected behavior here.

mumrah · 2023-08-31T14:08:33Z

+    // Unlike on the broker, DESCRIBE_CLUSTER on the controller requires a high level of
+    // permissions (ALTER on CLUSTER).


What's the reason behind this?

direct-to-controller operation is intended only for administrators. if you're not an admin, you should talk to the brokers. (In a well-run network, this should also be enforced by putting non-administrators on a separate subnet.)

Thanks for this excellent explanation, @cmccabe! I've created JIRA issue https://issues.apache.org/jira/browse/KAFKA-18434 to incorporate this explanation into the error message. This will help users understand the permission distinctions when using controllers as bootstrap servers.

mumrah · 2023-08-31T14:09:06Z

+import scala.jdk.CollectionConverters._
+
+/**
+ * The broker lifecycle manager owns the broker state.


This and other javadocs in this class need updating (looks like copy-pasta from BrokerLifecycleManager 🙃)

mumrah · 2023-08-31T14:12:11Z

+    manifest: LoaderManifest
+  ): Unit = {
+    if (delta.featuresDelta() != null ||
+        (delta.clusterDelta() != null && delta.clusterDelta().changedControllers().containsKey(nodeId))) {


Does the controller only care about its own registration in this class? Is that why we're filtering by nodeId here?

yes, it cares only about its own registration here

mumrah · 2023-08-31T14:14:44Z

+  /**
+   * The number of RPCs that we are waiting for. Only read or written from the event queue thread.
+   */
+  var pendingRpcs = 0L


Seems like we only allow one pending RPC at a time. Should we use a boolean here instead?

mumrah · 2023-08-31T14:18:23Z

+              registered = false
+            } else {
+              info(s"Our registration has been persisted to the metadata log.")
+              registered = true


Instead of registered, how about something like registeredInLog or something more specific. Since we're sending "Registration" RPCs, someone might interpret "registered" to mean that we've sent a registration request or something.

yeah. i'll change it to registeredInLog

mumrah · 2023-08-31T14:20:21Z


+      // Set up the controller registrations publisher.
+      metadataPublishers.add(controllerRegistrationsPublisher)
+


Does the registration manager need to be added to the metadataPublishers?

mumrah · 2023-08-31T14:21:48Z

-          .setHost(broker.host)
-          .setPort(broker.port)
-          .setRack(broker.rack))
+    val response = authHelper.computeDescribeClusterResponse(


Do we need to handle the case of someone requesting Controller endpoints from the broker (and vice versa)? Or is that handled somewhere else?

That case is handled in AuthHelper.computeDescribeClusterResponse

mumrah · 2023-08-31T14:23:31Z

+public interface ClusterSupportDescriber {
+    Iterator<Entry<Integer, Map<String, VersionRange>>> brokerSupported();
+    Iterator<Entry<Integer, Map<String, VersionRange>>> controllerSupported();
+}


nit: missing a blank line at bottom of file

mumrah · 2023-08-31T14:30:19Z

+import org.apache.kafka.metadata.VersionRange;
+
+
+public interface ClusterSupportDescriber {


How about ClusterFeatureDescriber, ClusterFeatureSupport, or something with "feature" in the name

ClusterFeatureSupportDescriber is fine, I guess. A bit long, but descriptive.

mumrah · 2023-08-31T14:34:03Z

+        int numBrokersChecked = 0;
+        int numControllersChecked = 0;
+        Optional<String> reason = quorumFeatures.reasonNotLocallySupported(featureName, newVersion);
+        if (reason.isPresent()) return reason;


nit: reformat this? (I don't think we normally do one-liner if statements?)

ehhh... we normally do allow them.

% git grep -w return | grep -w if | wc -l 1470

I think the rule is that you need braces for any if statement that spans multiple lines. Also if it gets (horizontally) long, you should cut that out.

We should probably teach checkstyle about this at some point, or at least write it down

cmccabe · 2023-08-31T23:01:29Z

One question I had was about de-registration (or lack thereof). How do we deal with one controller being offline? Do we just use its most recent registration?

We just use the most recent registration. Like you said, it's possible that this is out of date. But there's no getting around that in a distributed system.

More generally, we can't prevent people from bringing up nodes with old software. If you want to break your cluster by doing that, you can do it, and there's nothing we can do about it. We don't control the version of software deployed on your cluster.

The safety interlock is designed for the case where the admin is being "reasonable"

showuon · 2023-09-01T07:16:14Z

+     * @param context       The controller request context.
+     * @param request      The registration request.
+     *
+     * @return             A future yielding the broker registration reply.


nit: alignment

showuon · 2023-09-01T08:22:50Z

+    /*
+    @Test
+    public void testInitialControllers() {
+        ControllerRegistrationsPublisher publisher = new ControllerRegistrationsPublisher();
+        assertEquals(Collections.emptyMap(), publisher.controllers());
+    }
+


All commented out?

dengziming

Thanks for this change, I left 2 minor comments.

dengziming · 2023-09-04T12:08:52Z

+        private final ControllerRegistrationRequestData data;
+
+        public Builder(ControllerRegistrationRequestData data) {
+            super(ApiKeys.BROKER_HEARTBEAT);


this should be CONTROLLER_REGISTRATION

Nice catch, @dengziming !

Yes, nice catch. Fixed.

dengziming · 2023-09-04T12:13:43Z

+// limitations under the License.
+
+{
+  "apiKey": 70,


we don't have 69, why use 70?

edit: now using 70 again since someone claimed 69 :)

dengziming · 2023-09-04T12:52:44Z

    }

+    public short registerControllerRecordVersion() {
+        if (isAtLeast(MetadataVersion.IBP_3_6_IV2)) {


We are not �using IBP_3_7_IV0, does this mean a old-version controller can register a controller with an new-version controller?

Sorry, this was left over from when the feature was in IBP_3_6_IV2 previously. Fixed now to be IBP_3_7_IV0

showuon

LGTM!

cmccabe · 2023-09-07T17:13:14Z

Fixed conflicts

…llers (apache#14306) Implement KIP-919: Allow AdminClient to Talk Directly with the KRaft Controller Quorum and add Controller Registration. This KIP adds a new version of DescribeClusterRequest which is supported by KRaft controllers. It also teaches AdminClient how to use this new DESCRIBE_CLUSTER request to talk directly with the controller quorum. This is all gated behind a new MetadataVersion, IBP_3_7_IV0. In order to share the DESCRIBE_CLUSTER logic between broker and controller, this PR factors it out into AuthHelper.computeDescribeClusterResponse. The KIP adds three new errors codes: MISMATCHED_ENDPOINT_TYPE, UNSUPPORTED_ENDPOINT_TYPE, and UNKNOWN_CONTROLLER_ID. The endpoint type errors can be returned from DescribeClusterRequest On the controller side, the controllers now try to register themselves with the current active controller, by sending a CONTROLLER_REGISTRATION request. This, in turn, is converted into a RegisterControllerRecord by the active controller. ClusterImage, ClusterDelta, and all other associated classes have been upgraded to propagate the new metadata. In the metadata shell, the cluster directory now contains both broker and controller subdirectories. QuorumFeatures previously had a reference to the ApiVersions structure used by the controller's NetworkClient. Because this PR removes that reference, QuorumFeatures now contains only immutable data. Specifically, it contains the current node ID, the locally supported features, and the list of quorum node IDs in the cluster. Reviewers: David Arthur <mumrah@gmail.com>, Ziming Deng <dengziming1993@gmail.com>, Luke Chen <showuon@gmail.com>

bachmanity1 · 2024-08-20T19:18:54Z

+          case Some(value) =>
+            value match {
+              case ZkCachedControllerId (id) => id
+              case KRaftCachedControllerId (_) => metadataCache.getRandomAliveBrokerId.getOrElse(- 1)


Hi @cmccabe,
Thanks for the PR! Sorry for revisiting this, but since this PR allows clients to communicate directly with controllers, wouldn't it make more sense to return the active controller ID here instead of a random broker ID?

cmccabe added the kraft label Aug 29, 2023

cmccabe assigned mumrah Aug 29, 2023

showuon self-assigned this Aug 30, 2023

showuon reviewed Aug 31, 2023

View reviewed changes

Comment thread core/src/main/scala/kafka/server/AuthHelper.scala

showuon reviewed Aug 31, 2023

View reviewed changes

mumrah reviewed Aug 31, 2023

View reviewed changes

Revert change to core/src/test/resources/log4j.properties

b7f22fc

Fix ControllerRegistrationManager javadoc

3a994fc

showuon reviewed Sep 1, 2023

View reviewed changes

ControllerRegistrationManager: some refactors

064210d

dengziming reviewed Sep 4, 2023

View reviewed changes

cmccabe added 3 commits September 5, 2023 13:53

fix some issues related to rpc

6b8d99b

Merge branch 'trunk' into cmccabe_kip_919

c221189

fix merge

6625290

showuon approved these changes Sep 6, 2023

View reviewed changes

cmccabe added 7 commits September 6, 2023 11:03

Fix compile for scala 2.12

73e7ba7

Add end-to-end test of registration publishers working

d7f9012

Merge branch 'trunk' into cmccabe_kip_919

5b24e92

update MV

15aa2b9

test fixes

36a5f30

Fix more tests

d22a989

Merge branch 'trunk' into cmccabe_kip_919

9a73f7b

cmccabe merged commit 41b695b into apache:trunk Sep 7, 2023

cmccabe deleted the cmccabe_kip_919 branch September 7, 2023 22:22

dengziming mentioned this pull request Sep 19, 2023

MINOR: Fix NPE in controllerListeners from controller server #14394

Closed

3 tasks

rondagostino mentioned this pull request Oct 3, 2023

KAFKA-15355: Message schema changes #14290

Merged

3 tasks

bachmanity1 reviewed Aug 20, 2024

View reviewed changes

This was referenced Jan 8, 2025

KAFKA-18440: Admin does not convert the AuthorizationException to fatal error in using bootstrap controllers #18435

Merged

KAFKA-18434: enrich the authorization error message of connecting to controller #18436

Merged

		// Unlike on the broker, DESCRIBE_CLUSTER on the controller requires a high level of
		// permissions (ALTER on CLUSTER).


		// Set up the controller registrations publisher.
		metadataPublishers.add(controllerRegistrationsPublisher)

		import org.apache.kafka.metadata.VersionRange;


		public interface ClusterSupportDescriber {

Conversation

cmccabe commented Aug 29, 2023

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mumrah left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe Aug 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe Aug 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe commented Aug 31, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dengziming left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cmccabe Aug 31, 2023 •

edited

Loading

cmccabe Aug 31, 2023 •

edited

Loading

cmccabe Sep 5, 2023 •

edited

Loading

bachmanity1 Aug 20, 2024 •

edited

Loading