KAFKA-13883: Implement NoOpRecord and metadata metrics by jsancio · Pull Request #12183 · apache/kafka

jsancio · 2022-05-19T17:27:10Z

Public Changes

A new configuration flag metadata.max.idle.interval.ms
A new metrics group "broker-metadata-metrics" to the brokers
Four new metrics to the KafkaController metric group
Increase the IBP and metadata.version
A new metadata record called NoOpRecord which is only written if we have a matching metadata version

Implementation Changes

The KRaft controller schedules an event to write NoOpRecord to the metadata log if the metadata version supports this feature. This event is scheduled at the interval defined in metadata.max.idle.interval.ms. Brokers and controllers were improved to ignore the NoOpRecord when replaying the metadata log.

The replica control manager was also changed to accept a metadata version supplier to determine if leader recovery state is supported.

Other Changes

Fixed a bug where metadata version 3.3-IV1 was not marked as changing the metadata.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

cmccabe · 2022-05-25T23:02:30Z

          OptionalLong.empty()
        }

+        val maxIdleIntervalNs = config.metadataMaxIdleIntervalMs match {


It's more traditional to have conversion stuff like this in KafkaConfig. If you had a function like maxIdleIntervalNs there, it could return Option[Long] and we could convert that into OptionalLong as needed (I think a simple asScala can do it if you have the optional converter imported...)

Okay. I fixed this. asJava converts it to a Optional[Long] so I did it manually.

cmccabe · 2022-05-25T23:05:47Z

  val MetadataLogDirDoc = "This configuration determines where we put the metadata log for clusters in KRaft mode. " +
    "If it is not set, the metadata log is placed in the first log directory from log.dirs."
  val MetadataSnapshotMaxNewRecordBytesDoc = "This is the maximum number of bytes in the log between the latest snapshot and the high-watermark needed before generating a new snapshot."
+  val MetadataMaxIdleIntervalMsDoc = "This configuration controls how often the active " +


Can we disallow negative values for this? 0 for disabled seems clear enough.

cmccabe · 2022-05-25T23:06:46Z


  /************* Metadata Configuration ***********/
  val metadataSnapshotMaxNewRecordBytes = getLong(KafkaConfig.MetadataSnapshotMaxNewRecordBytesProp)
+  val metadataMaxIdleIntervalMs: Option[Int] = {


Like I wrote above, seems like the accessor we want should give us nanoseconds not milliseconds. I suppose the value in nanoseconds should be a long too...

cmccabe · 2022-05-25T23:07:07Z

      .define(MetadataLogSegmentMillisProp, LONG, Defaults.LogRollHours * 60 * 60 * 1000L, null, HIGH, MetadataLogSegmentMillisDoc)
      .define(MetadataMaxRetentionBytesProp, LONG, Defaults.LogRetentionBytes, null, HIGH, MetadataMaxRetentionBytesDoc)
      .define(MetadataMaxRetentionMillisProp, LONG, Defaults.LogRetentionHours * 60 * 60 * 1000L, null, HIGH, MetadataMaxRetentionMillisDoc)
+      .define(MetadataMaxIdleIntervalMsProp, INT, Defaults.MetadataMaxIdleIntervalMs, null, LOW, MetadataMaxIdleIntervalMsDoc)


can we valiate that this is not negative? like I wrote above, there seems to be no reason to allow negative values

cmccabe · 2022-05-25T23:09:14Z

+import org.apache.kafka.common.metrics.Metrics
+import org.apache.kafka.common.metrics.MetricConfig
+
+final class BrokerMetrics private (metrics: Metrics) extends AutoCloseable {


How do you feel about BrokerServerMetrics as a name for this? That might make it more clear that this doesn't apply to KafkaServer...

Sounds good to me.

cmccabe · 2022-05-25T23:10:32Z


    // Support for leader recovery for unclean leader election (KIP-704)
-    IBP_3_2_IV0(4, "3.2", "IV0", false),
+    IBP_3_2_IV0(4, "3.2", "IV0", true),


Thanks for fixing this. We'll have to make a note in the PR description as well

cmccabe · 2022-05-25T23:11:38Z


 import static java.util.concurrent.TimeUnit.NANOSECONDS;
-
+import java.util.ArrayList;


I kind of prefer the style where all the java.util imports are in a single block, can we do that here?

cmccabe · 2022-05-25T23:13:26Z

-            if (controllerMetrics == null) {
-                throw new RuntimeException("You must specify controllerMetrics.");
+            if (metadataVersion == null) {
+                throw new IllegalStateException("Metadata version must be set before building");


Let's just default to the latest version. It's very tedious to have to set this in every single unit test (most of which don't care)

Sounds good to me.

cmccabe · 2022-05-25T23:14:58Z

        private ReplicaPlacer replicaPlacer = new StripedReplicaPlacer(new Random());
        private long snapshotMaxNewRecordBytes = Long.MAX_VALUE;
        private OptionalLong leaderImbalanceCheckIntervalNs = OptionalLong.empty();
+        private OptionalLong maxIdleIntervalNs = OptionalLong.empty();


It's an interesting choice to have this off by default. I guess it probably would complicate a bunch of junit tests since it would create records we're not expecting. So maybe this is the correct way to go for now.

cmccabe · 2022-05-25T23:15:30Z

-                    leaderImbalanceCheckIntervalNs, sessionTimeoutNs, controllerMetrics,
-                    createTopicPolicy, alterConfigPolicy, configurationValidator, authorizer,
-                    staticConfig, bootstrapMetadata);
+                return new QuorumController(


sigh. all right, fine. I guess we might as well do one line per parameter.

Hehe. I thought it was improvement. I always find the previous style harder to read and update.

cmccabe · 2022-05-25T23:20:52Z

+            // the default append linger for KRaft is 25ms.
+            controllerMetrics.setLastAppliedRecordTimestamp(time.milliseconds());
+        } else {
+            // This is called with an offset of -1 when the active controller renounced leadership


Maybe the comment should just mention that lastAppliedRecordOffset tracks last committed offset for standbys.

The bit about writeOffset being -1 is a bit low-level (it is true but that feels more like an implementation detail here...)

…onitor-quorum

cmccabe

LGTM

jsancio · 2022-06-01T17:48:30Z

Merging since the failure is unrelated. The failure at in the ZK tests and this is a KRaft only change:

testTopicIdUpgradeAfterReassigningPartitions() – kafka.controller.ControllerIntegrationTest

KAFKA-13883: Implement NoOpRecord and metadata metrics

d94afda

jsancio force-pushed the kafka-13883-monitor-quorum branch from bbad90f to d94afda Compare May 19, 2022 18:19

jsancio added 6 commits May 20, 2022 08:37

Check NoOpRecord schedule when feature control changes

f89fb9b

Handle changes to feature control

d8f53c6

Support dynamic metadata version for leader recovery state

03d5bcd

Remove test TODO as it was implemented

93c014e

Close the metrics when closeing the listener

a64edb9

Fix typo

562404d

cmccabe reviewed May 25, 2022

View reviewed changes

jsancio added 4 commits May 26, 2022 11:07

Fix PR comments

bab8b0c

Merge remote-tracking branch 'apache-github/trunk' into kafka-13883-m…

1901d62

…onitor-quorum

Fix illegal cast exception

fb3fc9d

Only active controller should handle feature control change

5e43b7e

cmccabe approved these changes Jun 1, 2022

View reviewed changes

cmccabe merged commit 7d1b092 into apache:trunk Jun 1, 2022

jsancio deleted the kafka-13883-monitor-quorum branch June 1, 2022 17:48

dengziming mentioned this pull request Jun 2, 2022

KAFKA-13955: Fix failing KRaftClusterTest tests #12238

Merged

3 tasks


		import static java.util.concurrent.TimeUnit.NANOSECONDS;

		import java.util.ArrayList;

Conversation

jsancio commented May 19, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Public Changes

Implementation Changes

Other Changes

Committer Checklist (excluded from commit message)

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!

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 left a comment

Choose a reason for hiding this comment

Uh oh!

jsancio commented Jun 1, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jsancio commented May 19, 2022 •

edited

Loading

jsancio commented Jun 1, 2022 •

edited

Loading