From e411cccd0470e7f354030a6a47528c5d4d469575 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Tue, 20 Apr 2021 16:41:12 -0700 Subject: [PATCH 01/11] global-topic-count-metric-and-test --- .../kafka/controller/ControllerMetrics.java | 6 +++ .../kafka/controller/QuorumController.java | 2 +- .../controller/QuorumControllerMetrics.java | 27 +++++++++++ .../controller/ReplicationControlManager.java | 12 ++++- .../controller/MockControllerMetrics.java | 18 ++++++- .../ReplicationControlManagerTest.java | 48 ++++++++++++++++++- 6 files changed, 108 insertions(+), 5 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java index fd4f3befb805f..5cc7d4d0a9fe2 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java @@ -26,4 +26,10 @@ public interface ControllerMetrics { void updateEventQueueTime(long durationMs); void updateEventQueueProcessingTime(long durationMs); + + int topicCount(); + + void incTopicCount(); + + void decTopicCount(); } diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 86faa5ede8e6c..42bc308c24160 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -928,7 +928,7 @@ private QuorumController(LogContext logContext, this.snapshotGeneratorManager = new SnapshotGeneratorManager(snapshotWriterBuilder); this.replicationControl = new ReplicationControlManager(snapshotRegistry, logContext, defaultReplicationFactor, defaultNumPartitions, - configurationControl, clusterControl); + configurationControl, clusterControl, controllerMetrics); this.logManager = logManager; this.metaLogListener = new QuorumMetaLogListener(); this.curClaimEpoch = -1L; diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index ad56faf3da99c..2c442f2084238 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -30,11 +30,16 @@ public final class QuorumControllerMetrics implements ControllerMetrics { "kafka.controller", "ControllerEventManager", "EventQueueTimeMs", null); private final static MetricName EVENT_QUEUE_PROCESSING_TIME_MS = new MetricName( "kafka.controller", "ControllerEventManager", "EventQueueProcessingTimeMs", null); + private final static MetricName GLOBAL_TOPIC_COUNT = new MetricName( + "kafka.controller", "ReplicationControlManager", "GlobalTopicCount", null); + private volatile boolean active; private final Gauge activeControllerCount; private final Histogram eventQueueTime; private final Histogram eventQueueProcessingTime; + private int topicCount; + private final Gauge globalTopicCount; public QuorumControllerMetrics(MetricsRegistry registry) { this.active = false; @@ -46,6 +51,13 @@ public Integer value() { }); this.eventQueueTime = registry.newHistogram(EVENT_QUEUE_TIME_MS, true); this.eventQueueProcessingTime = registry.newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true); + this.topicCount = 0; + this.globalTopicCount = registry.newGauge(GLOBAL_TOPIC_COUNT, new Gauge() { + @Override + public Integer value() { + return topicCount; + } + }); } @Override @@ -67,4 +79,19 @@ public void updateEventQueueTime(long durationMs) { public void updateEventQueueProcessingTime(long durationMs) { eventQueueTime.update(durationMs); } + + @Override + public int topicCount() { + return topicCount; + } + + @Override + public void incTopicCount() { + topicCount++; + } + + @Override + public void decTopicCount() { + topicCount--; + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index f169d1fb4aafc..109df162ba414 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -274,6 +274,11 @@ public String toString() { */ private final ClusterControlManager clusterControl; + /** + * A reference to the controller's metrics registry. + */ + private final ControllerMetrics controllerMetrics; + /** * Maps topic names to topic UUIDs. */ @@ -294,12 +299,14 @@ public String toString() { short defaultReplicationFactor, int defaultNumPartitions, ConfigurationControlManager configurationControl, - ClusterControlManager clusterControl) { + ClusterControlManager clusterControl, + ControllerMetrics controllerMetrics) { this.snapshotRegistry = snapshotRegistry; this.log = logContext.logger(ReplicationControlManager.class); this.defaultReplicationFactor = defaultReplicationFactor; this.defaultNumPartitions = defaultNumPartitions; this.configurationControl = configurationControl; + this.controllerMetrics = controllerMetrics; this.clusterControl = clusterControl; this.topicsByName = new TimelineHashMap<>(snapshotRegistry, 0); this.topics = new TimelineHashMap<>(snapshotRegistry, 0); @@ -310,6 +317,7 @@ public void replay(TopicRecord record) { topicsByName.put(record.name(), record.topicId()); topics.put(record.topicId(), new TopicControlInfo(record.name(), snapshotRegistry, record.topicId())); + controllerMetrics.incTopicCount(); log.info("Created topic {} with ID {}.", record.name(), record.topicId()); } @@ -378,7 +386,7 @@ public void replay(RemoveTopicRecord record) { } } brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER); - + controllerMetrics.decTopicCount(); log.info("Removed topic {} with ID {}.", topic.name, record.topicId()); } diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index 4e6523e37f286..2c6e37f8fb800 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -17,12 +17,13 @@ package org.apache.kafka.controller; - public final class MockControllerMetrics implements ControllerMetrics { private volatile boolean active; + private int topicCount; public MockControllerMetrics() { this.active = false; + this.topicCount = 0; } @Override @@ -44,4 +45,19 @@ public void updateEventQueueTime(long durationMs) { public void updateEventQueueProcessingTime(long durationMs) { // nothing to do } + + @Override + public int topicCount() { + return this.topicCount; + } + + @Override + public void incTopicCount() { + this.topicCount++; + } + + @Override + public void decTopicCount() { + this.topicCount--; + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index e524581f32e36..85d85ba91d710 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -87,6 +87,7 @@ private static class ReplicationControlTestContext { final ClusterControlManager clusterControl = new ClusterControlManager( logContext, time, snapshotRegistry, 1000, new SimpleReplicaPlacementPolicy(random)); + final ControllerMetrics metrics = new MockControllerMetrics(); final ConfigurationControlManager configurationControl = new ConfigurationControlManager( new LogContext(), snapshotRegistry, Collections.emptyMap()); final ReplicationControlManager replicationControl = new ReplicationControlManager(snapshotRegistry, @@ -94,7 +95,8 @@ private static class ReplicationControlTestContext { (short) 3, 1, configurationControl, - clusterControl); + clusterControl, + metrics); void replay(List records) throws Exception { ControllerTestUtils.replayAll(clusterControl, records); @@ -181,6 +183,50 @@ public void testCreateTopics() throws Exception { ctx.replicationControl.iterator(Long.MAX_VALUE)); } + @Test + public void testGlobalTopicMetric() throws Exception { + ReplicationControlTestContext ctx = new ReplicationControlTestContext(); + ReplicationControlManager replicationControl = ctx.replicationControl; + CreateTopicsRequestData request = new CreateTopicsRequestData(); + request.topics().add(new CreatableTopic().setName("foo"). + setNumPartitions(-1).setReplicationFactor((short) -1)); + + registerBroker(0, ctx); + unfenceBroker(0, ctx); + registerBroker(1, ctx); + unfenceBroker(1, ctx); + registerBroker(2, ctx); + unfenceBroker(2, ctx); + + List topicsToDelete = new ArrayList<>(); + + ControllerResult result = + replicationControl.createTopics(request); + topicsToDelete.add(result.response().topics().find("foo").topicId()); + + ControllerTestUtils.replayAll(replicationControl, result.records()); + assertEquals(1, ctx.metrics.topicCount()); + + request = new CreateTopicsRequestData(); + request.topics().add(new CreatableTopic().setName("bar"). + setNumPartitions(-1).setReplicationFactor((short) -1)); + request.topics().add(new CreatableTopic().setName("baz"). + setNumPartitions(-1).setReplicationFactor((short) -1)); + result = replicationControl.createTopics(request); + ControllerTestUtils.replayAll(replicationControl, result.records()); + assertEquals(3, ctx.metrics.topicCount()); + + topicsToDelete.add(result.response().topics().find("baz").topicId()); + ControllerResult> deleteResult = replicationControl.deleteTopics(topicsToDelete); + ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); + assertEquals(1, ctx.metrics.topicCount()); + + Uuid topicToDelete = result.response().topics().find("bar").topicId(); + deleteResult = replicationControl.deleteTopics(Collections.singletonList(topicToDelete)); + ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); + assertEquals(0, ctx.metrics.topicCount()); + } + @Test public void testValidateNewTopicNames() { Map topicErrors = new HashMap<>(); From d1b502b7bf69a3b44c7f07867026d56b5c5eb9e3 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Tue, 20 Apr 2021 17:45:41 -0700 Subject: [PATCH 02/11] add-global-partition-and-topic-metrics --- .../kafka/controller/ControllerMetrics.java | 6 +++++ .../controller/QuorumControllerMetrics.java | 25 +++++++++++++++++++ .../controller/ReplicationControlManager.java | 2 ++ .../controller/MockControllerMetrics.java | 17 +++++++++++++ .../ReplicationControlManagerTest.java | 11 +++++--- 5 files changed, 57 insertions(+), 4 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java index 5cc7d4d0a9fe2..35e23fb6385bb 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java @@ -32,4 +32,10 @@ public interface ControllerMetrics { void incTopicCount(); void decTopicCount(); + + int partitionCount(); + + void incPartitionCount(); + + void decPartitionCount(); } diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index 2c442f2084238..07c5ec8e47e7d 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -32,6 +32,8 @@ public final class QuorumControllerMetrics implements ControllerMetrics { "kafka.controller", "ControllerEventManager", "EventQueueProcessingTimeMs", null); private final static MetricName GLOBAL_TOPIC_COUNT = new MetricName( "kafka.controller", "ReplicationControlManager", "GlobalTopicCount", null); + private final static MetricName GLOBAL_PARTITION_COUNT = new MetricName( + "kafka.controller", "ReplicationControlManager", "GlobalPartitionCount", null); private volatile boolean active; @@ -40,6 +42,8 @@ public final class QuorumControllerMetrics implements ControllerMetrics { private final Histogram eventQueueProcessingTime; private int topicCount; private final Gauge globalTopicCount; + private int partitionCount; + private final Gauge globalPartitionCount; public QuorumControllerMetrics(MetricsRegistry registry) { this.active = false; @@ -58,6 +62,12 @@ public Integer value() { return topicCount; } }); + this.globalPartitionCount = registry.newGauge(GLOBAL_TOPIC_COUNT, new Gauge() { + @Override + public Integer value() { + return partitionCount; + } + }); } @Override @@ -94,4 +104,19 @@ public void incTopicCount() { public void decTopicCount() { topicCount--; } + + @Override + public int partitionCount() { + return partitionCount; + } + + @Override + public void incPartitionCount() { + partitionCount++; + } + + @Override + public void decPartitionCount() { + partitionCount--; + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 109df162ba414..a6e83c07d77eb 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -346,6 +346,7 @@ public void replay(PartitionRecord record) { newPartInfo.leader); } } + controllerMetrics.incPartitionCount(); } public void replay(PartitionChangeRecord record) { @@ -384,6 +385,7 @@ public void replay(RemoveTopicRecord record) { for (int i = 0; i < partition.isr.length; i++) { brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]); } + controllerMetrics.decPartitionCount(); } brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER); controllerMetrics.decTopicCount(); diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index 2c6e37f8fb800..360df2f3b7c63 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -20,10 +20,12 @@ public final class MockControllerMetrics implements ControllerMetrics { private volatile boolean active; private int topicCount; + private int partitionCount; public MockControllerMetrics() { this.active = false; this.topicCount = 0; + this.partitionCount = 0; } @Override @@ -60,4 +62,19 @@ public void incTopicCount() { public void decTopicCount() { this.topicCount--; } + + @Override + public int partitionCount() { + return this.partitionCount; + } + + @Override + public void incPartitionCount() { + partitionCount++; + } + + @Override + public void decPartitionCount() { + partitionCount--; + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index 85d85ba91d710..ce756f8b38158 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -184,12 +184,12 @@ public void testCreateTopics() throws Exception { } @Test - public void testGlobalTopicMetric() throws Exception { + public void testGlobalTopicAndPartitionMetrics() throws Exception { ReplicationControlTestContext ctx = new ReplicationControlTestContext(); ReplicationControlManager replicationControl = ctx.replicationControl; CreateTopicsRequestData request = new CreateTopicsRequestData(); request.topics().add(new CreatableTopic().setName("foo"). - setNumPartitions(-1).setReplicationFactor((short) -1)); + setNumPartitions(1).setReplicationFactor((short) -1)); registerBroker(0, ctx); unfenceBroker(0, ctx); @@ -209,22 +209,25 @@ public void testGlobalTopicMetric() throws Exception { request = new CreateTopicsRequestData(); request.topics().add(new CreatableTopic().setName("bar"). - setNumPartitions(-1).setReplicationFactor((short) -1)); + setNumPartitions(1).setReplicationFactor((short) -1)); request.topics().add(new CreatableTopic().setName("baz"). - setNumPartitions(-1).setReplicationFactor((short) -1)); + setNumPartitions(2).setReplicationFactor((short) -1)); result = replicationControl.createTopics(request); ControllerTestUtils.replayAll(replicationControl, result.records()); assertEquals(3, ctx.metrics.topicCount()); + assertEquals(4, ctx.metrics.partitionCount()); topicsToDelete.add(result.response().topics().find("baz").topicId()); ControllerResult> deleteResult = replicationControl.deleteTopics(topicsToDelete); ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); assertEquals(1, ctx.metrics.topicCount()); + assertEquals(1, ctx.metrics.partitionCount()); Uuid topicToDelete = result.response().topics().find("bar").topicId(); deleteResult = replicationControl.deleteTopics(Collections.singletonList(topicToDelete)); ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); assertEquals(0, ctx.metrics.topicCount()); + assertEquals(0, ctx.metrics.partitionCount()); } @Test From d65e8e56a6f54d78baa108b59ffbeb90c6e7c709 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Wed, 21 Apr 2021 10:15:17 -0700 Subject: [PATCH 03/11] comments-addressed --- .../kafka/controller/ControllerMetrics.java | 12 +++---- .../controller/QuorumControllerMetrics.java | 36 +++++++------------ .../controller/ReplicationControlManager.java | 13 ++++--- .../controller/MockControllerMetrics.java | 28 +++++---------- .../ReplicationControlManagerTest.java | 14 ++++---- 5 files changed, 42 insertions(+), 61 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java index 35e23fb6385bb..406a533e7d7e4 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java @@ -27,15 +27,11 @@ public interface ControllerMetrics { void updateEventQueueProcessingTime(long durationMs); - int topicCount(); + void setGlobalTopicsCount(int topicCount); - void incTopicCount(); + int globalTopicsCount(); - void decTopicCount(); + void setGlobalPartitionCount(int partitionCount); - int partitionCount(); - - void incPartitionCount(); - - void decPartitionCount(); + int globalPartitionCount(); } diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index 07c5ec8e47e7d..3934ba3272aa6 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -37,13 +37,13 @@ public final class QuorumControllerMetrics implements ControllerMetrics { private volatile boolean active; + private volatile int topicCount; + private volatile int partitionCount; private final Gauge activeControllerCount; + private final Gauge globalPartitionCount; + private final Gauge globalTopicCount; private final Histogram eventQueueTime; private final Histogram eventQueueProcessingTime; - private int topicCount; - private final Gauge globalTopicCount; - private int partitionCount; - private final Gauge globalPartitionCount; public QuorumControllerMetrics(MetricsRegistry registry) { this.active = false; @@ -62,7 +62,7 @@ public Integer value() { return topicCount; } }); - this.globalPartitionCount = registry.newGauge(GLOBAL_TOPIC_COUNT, new Gauge() { + this.globalPartitionCount = registry.newGauge(GLOBAL_PARTITION_COUNT, new Gauge() { @Override public Integer value() { return partitionCount; @@ -91,32 +91,22 @@ public void updateEventQueueProcessingTime(long durationMs) { } @Override - public int topicCount() { - return topicCount; - } - - @Override - public void incTopicCount() { - topicCount++; - } - - @Override - public void decTopicCount() { - topicCount--; + public void setGlobalTopicsCount(int topicCount) { + this.topicCount = topicCount; } @Override - public int partitionCount() { - return partitionCount; + public int globalTopicsCount() { + return this.topicCount; } @Override - public void incPartitionCount() { - partitionCount++; + public void setGlobalPartitionCount(int partitionCount) { + this.partitionCount = partitionCount; } @Override - public void decPartitionCount() { - partitionCount--; + public int globalPartitionCount() { + return this.partitionCount; } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index a6e83c07d77eb..339e2c1f4c225 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -264,6 +264,8 @@ public String toString() { */ private final int defaultNumPartitions; + private int globalPartitionCount; + /** * A reference to the controller's configuration control manager. */ @@ -308,6 +310,7 @@ public String toString() { this.configurationControl = configurationControl; this.controllerMetrics = controllerMetrics; this.clusterControl = clusterControl; + this.globalPartitionCount = 0; this.topicsByName = new TimelineHashMap<>(snapshotRegistry, 0); this.topics = new TimelineHashMap<>(snapshotRegistry, 0); this.brokersToIsrs = new BrokersToIsrs(snapshotRegistry); @@ -317,7 +320,7 @@ public void replay(TopicRecord record) { topicsByName.put(record.name(), record.topicId()); topics.put(record.topicId(), new TopicControlInfo(record.name(), snapshotRegistry, record.topicId())); - controllerMetrics.incTopicCount(); + controllerMetrics.setGlobalTopicsCount(topics.size()); log.info("Created topic {} with ID {}.", record.name(), record.topicId()); } @@ -335,6 +338,7 @@ public void replay(PartitionRecord record) { topicInfo.parts.put(record.partitionId(), newPartInfo); brokersToIsrs.update(record.topicId(), record.partitionId(), null, newPartInfo.isr, NO_LEADER, newPartInfo.leader); + globalPartitionCount++; } else { String diff = newPartInfo.diff(prevPartInfo); if (!diff.isEmpty()) { @@ -346,7 +350,7 @@ public void replay(PartitionRecord record) { newPartInfo.leader); } } - controllerMetrics.incPartitionCount(); + controllerMetrics.setGlobalPartitionCount(globalPartitionCount); } public void replay(PartitionChangeRecord record) { @@ -385,10 +389,11 @@ public void replay(RemoveTopicRecord record) { for (int i = 0; i < partition.isr.length; i++) { brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]); } - controllerMetrics.decPartitionCount(); + globalPartitionCount--; } brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER); - controllerMetrics.decTopicCount(); + controllerMetrics.setGlobalTopicsCount(topics.size()); + controllerMetrics.setGlobalPartitionCount(globalPartitionCount); log.info("Removed topic {} with ID {}.", topic.name, record.topicId()); } diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index 360df2f3b7c63..10cf217ebb087 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -19,8 +19,8 @@ public final class MockControllerMetrics implements ControllerMetrics { private volatile boolean active; - private int topicCount; - private int partitionCount; + private volatile int topicCount; + private volatile int partitionCount; public MockControllerMetrics() { this.active = false; @@ -49,32 +49,22 @@ public void updateEventQueueProcessingTime(long durationMs) { } @Override - public int topicCount() { - return this.topicCount; + public void setGlobalTopicsCount(int topicCount) { + this.topicCount = topicCount; } @Override - public void incTopicCount() { - this.topicCount++; + public int globalTopicsCount() { + return this.topicCount; } @Override - public void decTopicCount() { - this.topicCount--; + public void setGlobalPartitionCount(int partitionCount) { + this.partitionCount = partitionCount; } @Override - public int partitionCount() { + public int globalPartitionCount() { return this.partitionCount; } - - @Override - public void incPartitionCount() { - partitionCount++; - } - - @Override - public void decPartitionCount() { - partitionCount--; - } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index ce756f8b38158..be75ed5e1c48b 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -205,7 +205,7 @@ public void testGlobalTopicAndPartitionMetrics() throws Exception { topicsToDelete.add(result.response().topics().find("foo").topicId()); ControllerTestUtils.replayAll(replicationControl, result.records()); - assertEquals(1, ctx.metrics.topicCount()); + assertEquals(1, ctx.metrics.globalTopicsCount()); request = new CreateTopicsRequestData(); request.topics().add(new CreatableTopic().setName("bar"). @@ -214,20 +214,20 @@ public void testGlobalTopicAndPartitionMetrics() throws Exception { setNumPartitions(2).setReplicationFactor((short) -1)); result = replicationControl.createTopics(request); ControllerTestUtils.replayAll(replicationControl, result.records()); - assertEquals(3, ctx.metrics.topicCount()); - assertEquals(4, ctx.metrics.partitionCount()); + assertEquals(3, ctx.metrics.globalTopicsCount()); + assertEquals(4, ctx.metrics.globalPartitionCount()); topicsToDelete.add(result.response().topics().find("baz").topicId()); ControllerResult> deleteResult = replicationControl.deleteTopics(topicsToDelete); ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); - assertEquals(1, ctx.metrics.topicCount()); - assertEquals(1, ctx.metrics.partitionCount()); + assertEquals(1, ctx.metrics.globalTopicsCount()); + assertEquals(1, ctx.metrics.globalPartitionCount()); Uuid topicToDelete = result.response().topics().find("bar").topicId(); deleteResult = replicationControl.deleteTopics(Collections.singletonList(topicToDelete)); ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); - assertEquals(0, ctx.metrics.topicCount()); - assertEquals(0, ctx.metrics.partitionCount()); + assertEquals(0, ctx.metrics.globalTopicsCount()); + assertEquals(0, ctx.metrics.globalPartitionCount()); } @Test From b21974c528bcea2552d3c8026245191ce27ea11a Mon Sep 17 00:00:00 2001 From: dielhennr Date: Mon, 10 May 2021 16:58:53 -0700 Subject: [PATCH 04/11] adding-offline-partition-and-replica-imbalance-metrics --- .../kafka/controller/BrokersToIsrs.java | 10 +++ .../kafka/controller/ControllerMetrics.java | 8 +++ .../controller/QuorumControllerMetrics.java | 61 ++++++++++++++++--- .../controller/ReplicationControlManager.java | 20 +++++- .../controller/MockControllerMetrics.java | 40 +++++++++--- .../ReplicationControlManagerTest.java | 54 ++++++++++++++++ 6 files changed, 175 insertions(+), 18 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java index d8e0319f94f7e..441ee32eb8c92 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java +++ b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java @@ -326,4 +326,14 @@ PartitionsOnReplicaIterator partitionsWithBrokerInIsr(int brokerId) { boolean hasLeaderships(int brokerId) { return iterator(brokerId, true).hasNext(); } + + int offlinePartitionCount() { + PartitionsOnReplicaIterator noLeaderIterator = partitionsWithNoLeader(); + int offlinePartitionCount = 0; + while (noLeaderIterator.hasNext()) { + noLeaderIterator.next(); + offlinePartitionCount++; + } + return offlinePartitionCount; + } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java index 406a533e7d7e4..7c862bc33c059 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java @@ -34,4 +34,12 @@ public interface ControllerMetrics { void setGlobalPartitionCount(int partitionCount); int globalPartitionCount(); + + void setOfflinePartitionCount(int offlinePartitions); + + int offlinePartitionCount(); + + void setPreferredReplicaImbalanceCount(int replicaImbalances); + + int preferredReplicaImbalanceCount(); } diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index 3934ba3272aa6..c33c383ffc972 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -34,19 +34,31 @@ public final class QuorumControllerMetrics implements ControllerMetrics { "kafka.controller", "ReplicationControlManager", "GlobalTopicCount", null); private final static MetricName GLOBAL_PARTITION_COUNT = new MetricName( "kafka.controller", "ReplicationControlManager", "GlobalPartitionCount", null); + private final static MetricName OFFLINE_PARTITION_COUNT = new MetricName( + "kafka.controller", "ReplicationControlManager", "OfflinePartitionCount", null); + private final static MetricName PREFERRED_REPLICA_IMBALANCE_COUNT = new MetricName( + "kafka.controller", "ReplicationControlManager", "PreferredReplicaImbalanceCount", null); private volatile boolean active; - private volatile int topicCount; - private volatile int partitionCount; + private volatile int topics; + private volatile int partitions; + private volatile int offlinePartitions; + private volatile int preferredReplicaImbalances; private final Gauge activeControllerCount; private final Gauge globalPartitionCount; private final Gauge globalTopicCount; + private final Gauge offlinePartitionCount; + private final Gauge preferredReplicaImbalanceCount; private final Histogram eventQueueTime; private final Histogram eventQueueProcessingTime; public QuorumControllerMetrics(MetricsRegistry registry) { this.active = false; + this.topics = 0; + this.partitions = 0; + this.offlinePartitions = 0; + this.preferredReplicaImbalances = 0; this.activeControllerCount = registry.newGauge(ACTIVE_CONTROLLER_COUNT, new Gauge() { @Override public Integer value() { @@ -55,17 +67,28 @@ public Integer value() { }); this.eventQueueTime = registry.newHistogram(EVENT_QUEUE_TIME_MS, true); this.eventQueueProcessingTime = registry.newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true); - this.topicCount = 0; this.globalTopicCount = registry.newGauge(GLOBAL_TOPIC_COUNT, new Gauge() { @Override public Integer value() { - return topicCount; + return topics; } }); this.globalPartitionCount = registry.newGauge(GLOBAL_PARTITION_COUNT, new Gauge() { @Override public Integer value() { - return partitionCount; + return partitions; + } + }); + this.offlinePartitionCount = registry.newGauge(OFFLINE_PARTITION_COUNT, new Gauge() { + @Override + public Integer value() { + return offlinePartitions; + } + }); + this.preferredReplicaImbalanceCount = registry.newGauge(PREFERRED_REPLICA_IMBALANCE_COUNT, new Gauge() { + @Override + public Integer value() { + return preferredReplicaImbalances; } }); } @@ -92,21 +115,41 @@ public void updateEventQueueProcessingTime(long durationMs) { @Override public void setGlobalTopicsCount(int topicCount) { - this.topicCount = topicCount; + this.topics = topicCount; } @Override public int globalTopicsCount() { - return this.topicCount; + return this.topics; } @Override public void setGlobalPartitionCount(int partitionCount) { - this.partitionCount = partitionCount; + this.partitions = partitionCount; } @Override public int globalPartitionCount() { - return this.partitionCount; + return this.partitions; + } + + @Override + public void setOfflinePartitionCount(int offlinePartitions) { + this.offlinePartitions = offlinePartitions; + } + + @Override + public int offlinePartitionCount() { + return this.offlinePartitions; + } + + @Override + public void setPreferredReplicaImbalanceCount(int replicaImbalances) { + this.preferredReplicaImbalances = replicaImbalances; + } + + @Override + public int preferredReplicaImbalanceCount() { + return this.preferredReplicaImbalances; } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 113bcf397fc31..ed7994dd41f23 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -354,14 +354,16 @@ public void replay(PartitionRecord record) { topicInfo.parts.put(record.partitionId(), newPartInfo); brokersToIsrs.update(record.topicId(), record.partitionId(), null, newPartInfo.isr, NO_LEADER, newPartInfo.leader); - } else if (!newPartInfo.equals(prevPartInfo)) { globalPartitionCount++; + } else if (!newPartInfo.equals(prevPartInfo)) { newPartInfo.maybeLogPartitionChange(log, description, prevPartInfo); topicInfo.parts.put(record.partitionId(), newPartInfo); brokersToIsrs.update(record.topicId(), record.partitionId(), prevPartInfo.isr, newPartInfo.isr, prevPartInfo.leader, newPartInfo.leader); } controllerMetrics.setGlobalPartitionCount(globalPartitionCount); + controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount()); } public void replay(PartitionChangeRecord record) { @@ -383,6 +385,8 @@ public void replay(PartitionChangeRecord record) { String topicPart = topicInfo.name + "-" + record.partitionId() + " with topic ID " + record.topicId(); newPartitionInfo.maybeLogPartitionChange(log, topicPart, prevPartitionInfo); + controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount()); } public void replay(RemoveTopicRecord record) { @@ -407,6 +411,8 @@ public void replay(RemoveTopicRecord record) { brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER); controllerMetrics.setGlobalTopicsCount(topics.size()); controllerMetrics.setGlobalPartitionCount(globalPartitionCount); + controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount()); log.info("Removed topic {} with ID {}.", topic.name, record.topicId()); } @@ -471,6 +477,18 @@ public void replay(RemoveTopicRecord record) { return ControllerResult.atomicOf(records, data); } + private int preferredReplicaImbalanceCount() { + int count = 0; + for (TopicControlInfo topic : topics.values()) { + for (PartitionControlInfo part : topic.parts.values()) { + if (part.leader != part.preferredReplica()) { + count++; + } + } + } + return count; + } + private ApiError createTopic(CreatableTopic topic, List records, Map successes) { diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index 10cf217ebb087..f13b24865a637 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -19,13 +19,17 @@ public final class MockControllerMetrics implements ControllerMetrics { private volatile boolean active; - private volatile int topicCount; - private volatile int partitionCount; + private volatile int topics; + private volatile int partitions; + private volatile int offlinePartitions; + private volatile int preferredReplicaImbalances; public MockControllerMetrics() { this.active = false; - this.topicCount = 0; - this.partitionCount = 0; + this.topics = 0; + this.partitions = 0; + this.offlinePartitions = 0; + this.preferredReplicaImbalances = 0; } @Override @@ -50,21 +54,41 @@ public void updateEventQueueProcessingTime(long durationMs) { @Override public void setGlobalTopicsCount(int topicCount) { - this.topicCount = topicCount; + this.topics = topicCount; } @Override public int globalTopicsCount() { - return this.topicCount; + return this.topics; } @Override public void setGlobalPartitionCount(int partitionCount) { - this.partitionCount = partitionCount; + this.partitions = partitionCount; } @Override public int globalPartitionCount() { - return this.partitionCount; + return this.partitions; + } + + @Override + public void setOfflinePartitionCount(int offlinePartitions) { + this.offlinePartitions = offlinePartitions; + } + + @Override + public int offlinePartitionCount() { + return this.offlinePartitions; + } + + @Override + public void setPreferredReplicaImbalanceCount(int replicaImbalances) { + this.preferredReplicaImbalances = replicaImbalances; + } + + @Override + public int preferredReplicaImbalanceCount() { + return this.preferredReplicaImbalances; } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index 38d2cd9cd5fcf..badd1f86dd73e 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -252,6 +252,60 @@ public void testGlobalTopicAndPartitionMetrics() throws Exception { assertEquals(0, ctx.metrics.globalPartitionCount()); } + @Test + public void testOfflinePartitionAndReplicaImbalanceMetrics() throws Exception { + ReplicationControlTestContext ctx = new ReplicationControlTestContext(); + ReplicationControlManager replicationControl = ctx.replicationControl; + + for (int i = 0; i < 4; i++) { + registerBroker(i, ctx); + unfenceBroker(i, ctx); + } + + ctx.createTestTopic("foo", + new int[][] { + new int[] {0, 2}, + new int[] {0, 1} + }); + + ctx.createTestTopic("zar", + new int[][] { + new int[] {0, 1, 2}, + new int[] {1, 2, 3}, + new int[] {1, 2, 0} + }); + + ControllerResult result = replicationControl.unregisterBroker(0); + ctx.replay(result.records()); + + // All partitions should still be online after unregistering broker 0 + assertEquals(0, ctx.metrics.offlinePartitionCount()); + // Three partitions should not have their preferred (first) replica 0 + assertEquals(3, ctx.metrics.preferredReplicaImbalanceCount()); + + result = replicationControl.unregisterBroker(1); + ctx.replay(result.records()); + + // After unregistering broker 1, 1 partition for topic foo should go offline + assertEquals(1, ctx.metrics.offlinePartitionCount()); + // All five partitions should not have their preferred (first) replica at this point + assertEquals(5, ctx.metrics.preferredReplicaImbalanceCount()); + + result = replicationControl.unregisterBroker(2); + ctx.replay(result.records()); + + // After unregistering broker 2, the last partition for topic foo should go offline + // and 2 partitions for topic zar should go offline + assertEquals(4, ctx.metrics.offlinePartitionCount()); + + result = replicationControl.unregisterBroker(3); + ctx.replay(result.records()); + + // After unregistering broker 3 the last partition for topic zar should go offline + assertEquals(5, ctx.metrics.offlinePartitionCount()); + } + + @Test public void testValidateNewTopicNames() { Map topicErrors = new HashMap<>(); From c35c00527ece881f7b4e1251b4ab822fbac9d9b4 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Wed, 12 May 2021 14:40:55 -0700 Subject: [PATCH 05/11] address-comments --- .../kafka/controller/BrokersToIsrs.java | 28 ++++++-- .../kafka/controller/ControllerMetrics.java | 8 +-- .../kafka/controller/QuorumController.java | 10 ++- .../controller/QuorumControllerMetrics.java | 43 ++++-------- .../controller/ReplicationControlManager.java | 39 +++++------ .../apache/kafka/queue/KafkaEventQueue.java | 14 ++++ .../controller/MockControllerMetrics.java | 24 ++----- .../ReplicationControlManagerTest.java | 65 +++++-------------- 8 files changed, 96 insertions(+), 135 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java index 441ee32eb8c92..12b798d7d180b 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java +++ b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java @@ -24,6 +24,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.Iterator; +import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; import java.util.NoSuchElementException; @@ -139,10 +140,16 @@ public TopicIdPartition next() { * Partitions with no isr members appear in this map under id NO_LEADER. */ private final TimelineHashMap> isrMembers; + + private final Map offlinePartitionCounts; + + private int offlinePartitionCount; BrokersToIsrs(SnapshotRegistry snapshotRegistry) { this.snapshotRegistry = snapshotRegistry; this.isrMembers = new TimelineHashMap<>(snapshotRegistry, 0); + this.offlinePartitionCounts = new HashMap<>(); + this.offlinePartitionCount = 0; } /** @@ -163,6 +170,11 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, } else { if (prevLeader == NO_LEADER) { prev = Replicas.copyWith(prevIsr, NO_LEADER); + if (nextLeader != NO_LEADER) { + int offlinePartitionsForTopic = offlinePartitionCounts.getOrDefault(topicId, 0); + offlinePartitionCounts.put(topicId, --offlinePartitionsForTopic); + offlinePartitionCount--; + } } else { prev = Replicas.clone(prevIsr); } @@ -174,6 +186,11 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, } else { if (nextLeader == NO_LEADER) { next = Replicas.copyWith(nextIsr, NO_LEADER); + if (prevLeader != NO_LEADER) { + int offlinePartitionsForTopic = offlinePartitionCounts.getOrDefault(topicId, 0); + offlinePartitionCounts.put(topicId, ++offlinePartitionsForTopic); + offlinePartitionCount++; + } } else { next = Replicas.clone(nextIsr); } @@ -221,6 +238,11 @@ void removeTopicEntryForBroker(Uuid topicId, int brokerId) { } } + void updateMetricsForTopicRemoval(Uuid topicId) { + offlinePartitionCount = offlinePartitionCount - offlinePartitionCounts.getOrDefault(topicId, 0); + offlinePartitionCounts.put(topicId, 0); + } + private void add(int brokerId, Uuid topicId, int newPartition, boolean leader) { if (leader) { newPartition = newPartition | LEADER_FLAG; @@ -328,12 +350,6 @@ boolean hasLeaderships(int brokerId) { } int offlinePartitionCount() { - PartitionsOnReplicaIterator noLeaderIterator = partitionsWithNoLeader(); - int offlinePartitionCount = 0; - while (noLeaderIterator.hasNext()) { - noLeaderIterator.next(); - offlinePartitionCount++; - } return offlinePartitionCount; } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java index 7c862bc33c059..afb06862ca3af 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java @@ -27,13 +27,9 @@ public interface ControllerMetrics { void updateEventQueueProcessingTime(long durationMs); - void setGlobalTopicsCount(int topicCount); + void setEventQueueSize(int queueSize); - int globalTopicsCount(); - - void setGlobalPartitionCount(int partitionCount); - - int globalPartitionCount(); + int eventQueueSize(); void setOfflinePartitionCount(int offlinePartitions); diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 42bc308c24160..38499a865326a 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -331,6 +331,7 @@ public String toString() { private void appendControlEvent(String name, Runnable handler) { ControlEvent event = new ControlEvent(name, handler); queue.append(event); + controllerMetrics.setEventQueueSize(queue.size()); } private static final String GENERATE_SNAPSHOT = "generateSnapshot"; @@ -374,12 +375,14 @@ void cancel() { generator.writer().close(); generator = null; queue.cancelDeferred(GENERATE_SNAPSHOT); + controllerMetrics.setEventQueueSize(queue.size()); } void reschedule(long delayNs) { ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this); queue.scheduleDeferred(event.name, new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event); + controllerMetrics.setEventQueueSize(queue.size()); } @Override @@ -472,6 +475,7 @@ ReplicationControlManager replicationControl() { CompletableFuture appendReadEvent(String name, Supplier handler) { ControllerReadEvent event = new ControllerReadEvent(name, handler); queue.append(event); + controllerMetrics.setEventQueueSize(queue.size()); return event.future(); } @@ -607,6 +611,7 @@ private CompletableFuture appendWriteEvent(String name, ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); queue.appendWithDeadline(time.nanoseconds() + NANOSECONDS.convert(timeoutMs, TimeUnit.MILLISECONDS), event); + controllerMetrics.setEventQueueSize(queue.size()); return event.future(); } @@ -614,6 +619,7 @@ private CompletableFuture appendWriteEvent(String name, ControllerWriteOperation op) { ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); queue.append(event); + controllerMetrics.setEventQueueSize(queue.size()); return event.future(); } @@ -708,6 +714,7 @@ private void scheduleDeferredWriteEvent(String name, long deadlineNs, ControllerWriteOperation op) { ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); queue.scheduleDeferred(name, new EarliestDeadlineFunction(deadlineNs), event); + controllerMetrics.setEventQueueSize(queue.size()); event.future.exceptionally(e -> { if (e instanceof UnknownServerException && e.getCause() != null && e.getCause() instanceof RejectedExecutionException) { @@ -744,6 +751,7 @@ private void rescheduleMaybeFenceStaleBrokers() { private void cancelMaybeFenceReplicas() { queue.cancelDeferred(MAYBE_FENCE_REPLICAS); + controllerMetrics.setEventQueueSize(queue.size()); } @SuppressWarnings("unchecked") @@ -821,7 +829,7 @@ private void replay(ApiMessage message, long snapshotEpoch, long offset) { /** * The controller metrics. */ - private final ControllerMetrics controllerMetrics; + final ControllerMetrics controllerMetrics; /** * A registry for snapshot data. This must be accessed only by the event queue thread. diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index c33c383ffc972..3fc8975fc6697 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -30,10 +30,8 @@ public final class QuorumControllerMetrics implements ControllerMetrics { "kafka.controller", "ControllerEventManager", "EventQueueTimeMs", null); private final static MetricName EVENT_QUEUE_PROCESSING_TIME_MS = new MetricName( "kafka.controller", "ControllerEventManager", "EventQueueProcessingTimeMs", null); - private final static MetricName GLOBAL_TOPIC_COUNT = new MetricName( - "kafka.controller", "ReplicationControlManager", "GlobalTopicCount", null); - private final static MetricName GLOBAL_PARTITION_COUNT = new MetricName( - "kafka.controller", "ReplicationControlManager", "GlobalPartitionCount", null); + private final static MetricName EVENT_QUEUE_SIZE = new MetricName( + "kafka.controller", "ControllerEventManager", "EventQueueSize", null); private final static MetricName OFFLINE_PARTITION_COUNT = new MetricName( "kafka.controller", "ReplicationControlManager", "OfflinePartitionCount", null); private final static MetricName PREFERRED_REPLICA_IMBALANCE_COUNT = new MetricName( @@ -41,24 +39,21 @@ public final class QuorumControllerMetrics implements ControllerMetrics { private volatile boolean active; - private volatile int topics; - private volatile int partitions; private volatile int offlinePartitions; private volatile int preferredReplicaImbalances; + private volatile int queueSize; private final Gauge activeControllerCount; - private final Gauge globalPartitionCount; - private final Gauge globalTopicCount; private final Gauge offlinePartitionCount; private final Gauge preferredReplicaImbalanceCount; + private final Gauge eventQueueSize; private final Histogram eventQueueTime; private final Histogram eventQueueProcessingTime; public QuorumControllerMetrics(MetricsRegistry registry) { this.active = false; - this.topics = 0; - this.partitions = 0; this.offlinePartitions = 0; this.preferredReplicaImbalances = 0; + this.queueSize = 0; this.activeControllerCount = registry.newGauge(ACTIVE_CONTROLLER_COUNT, new Gauge() { @Override public Integer value() { @@ -67,16 +62,10 @@ public Integer value() { }); this.eventQueueTime = registry.newHistogram(EVENT_QUEUE_TIME_MS, true); this.eventQueueProcessingTime = registry.newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true); - this.globalTopicCount = registry.newGauge(GLOBAL_TOPIC_COUNT, new Gauge() { + this.eventQueueSize = registry.newGauge(EVENT_QUEUE_SIZE, new Gauge() { @Override public Integer value() { - return topics; - } - }); - this.globalPartitionCount = registry.newGauge(GLOBAL_PARTITION_COUNT, new Gauge() { - @Override - public Integer value() { - return partitions; + return queueSize; } }); this.offlinePartitionCount = registry.newGauge(OFFLINE_PARTITION_COUNT, new Gauge() { @@ -114,23 +103,13 @@ public void updateEventQueueProcessingTime(long durationMs) { } @Override - public void setGlobalTopicsCount(int topicCount) { - this.topics = topicCount; - } - - @Override - public int globalTopicsCount() { - return this.topics; - } - - @Override - public void setGlobalPartitionCount(int partitionCount) { - this.partitions = partitionCount; + public void setEventQueueSize(int queueSize) { + this.queueSize = queueSize; } @Override - public int globalPartitionCount() { - return this.partitions; + public int eventQueueSize() { + return this.queueSize; } @Override diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index ed7994dd41f23..640d77f5298df 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -279,7 +279,7 @@ public String toString() { */ private final int defaultNumPartitions; - private int globalPartitionCount; + private int preferredReplicaImbalanceCount; /** * A reference to the controller's configuration control manager. @@ -325,7 +325,7 @@ public String toString() { this.configurationControl = configurationControl; this.controllerMetrics = controllerMetrics; this.clusterControl = clusterControl; - this.globalPartitionCount = 0; + this.preferredReplicaImbalanceCount = 0; this.topicsByName = new TimelineHashMap<>(snapshotRegistry, 0); this.topics = new TimelineHashMap<>(snapshotRegistry, 0); this.brokersToIsrs = new BrokersToIsrs(snapshotRegistry); @@ -335,7 +335,6 @@ public void replay(TopicRecord record) { topicsByName.put(record.name(), record.topicId()); topics.put(record.topicId(), new TopicControlInfo(record.name(), snapshotRegistry, record.topicId())); - controllerMetrics.setGlobalTopicsCount(topics.size()); log.info("Created topic {} with topic ID {}.", record.name(), record.topicId()); } @@ -354,16 +353,17 @@ public void replay(PartitionRecord record) { topicInfo.parts.put(record.partitionId(), newPartInfo); brokersToIsrs.update(record.topicId(), record.partitionId(), null, newPartInfo.isr, NO_LEADER, newPartInfo.leader); - globalPartitionCount++; } else if (!newPartInfo.equals(prevPartInfo)) { newPartInfo.maybeLogPartitionChange(log, description, prevPartInfo); topicInfo.parts.put(record.partitionId(), newPartInfo); brokersToIsrs.update(record.topicId(), record.partitionId(), prevPartInfo.isr, newPartInfo.isr, prevPartInfo.leader, newPartInfo.leader); } - controllerMetrics.setGlobalPartitionCount(globalPartitionCount); + if (newPartInfo.leader != newPartInfo.preferredReplica()) { + preferredReplicaImbalanceCount++; + } controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); - controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount()); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount); } public void replay(PartitionChangeRecord record) { @@ -385,8 +385,12 @@ public void replay(PartitionChangeRecord record) { String topicPart = topicInfo.name + "-" + record.partitionId() + " with topic ID " + record.topicId(); newPartitionInfo.maybeLogPartitionChange(log, topicPart, prevPartitionInfo); + if ((newPartitionInfo.leader != newPartitionInfo.preferredReplica()) && + (prevPartitionInfo.leader == prevPartitionInfo.preferredReplica())) { + preferredReplicaImbalanceCount++; + } controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); - controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount()); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount); } public void replay(RemoveTopicRecord record) { @@ -406,13 +410,14 @@ public void replay(RemoveTopicRecord record) { for (int i = 0; i < partition.isr.length; i++) { brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]); } - globalPartitionCount--; + if (partition.leader != partition.preferredReplica()) { + preferredReplicaImbalanceCount--; + } } + brokersToIsrs.updateMetricsForTopicRemoval(topic.id); brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER); - controllerMetrics.setGlobalTopicsCount(topics.size()); - controllerMetrics.setGlobalPartitionCount(globalPartitionCount); controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); - controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount()); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount); log.info("Removed topic {} with ID {}.", topic.name, record.topicId()); } @@ -477,18 +482,6 @@ public void replay(RemoveTopicRecord record) { return ControllerResult.atomicOf(records, data); } - private int preferredReplicaImbalanceCount() { - int count = 0; - for (TopicControlInfo topic : topics.values()) { - for (PartitionControlInfo part : topic.parts.values()) { - if (part.leader != part.preferredReplica()) { - count++; - } - } - } - return count; - } - private ApiError createTopic(CreatableTopic topic, List records, Map successes) { diff --git a/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java b/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java index 05642f6bac8b5..209ee19e3e363 100644 --- a/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java +++ b/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java @@ -157,6 +157,8 @@ private class EventHandler implements Runnable { */ private final EventContext head = new EventContext(null, null, null); + private int queueSize = 0; + /** * An ordered map of times in monotonic nanoseconds to events to time out. */ @@ -177,8 +179,14 @@ public void run() { } } + public int queueSize() { + return this.queueSize; + } + private void remove(EventContext eventContext) { eventContext.remove(); + System.out.println("removing" + eventContext.toString()); + this.queueSize--; if (eventContext.deadlineNs.isPresent()) { deadlineMap.remove(eventContext.deadlineNs.getAsLong()); eventContext.deadlineNs = OptionalLong.empty(); @@ -277,6 +285,8 @@ Exception enqueue(EventContext eventContext, OptionalLong deadlineNs = deadlineNsCalculator.apply(existingDeadlineNs); boolean queueWasEmpty = head.isSingleton(); boolean shouldSignal = false; + queueSize++; + System.out.println(eventContext.toString() + eventContext.insertionType.toString() + " " + queueSize); switch (eventContext.insertionType) { case APPEND: head.insertBefore(eventContext); @@ -370,6 +380,10 @@ public KafkaEventQueue(Time time, this.eventHandlerThread.start(); } + public int size() { + return this.eventHandler.queueSize(); + } + @Override public void enqueue(EventInsertionType insertionType, String tag, diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index f13b24865a637..2412bd388fce6 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -19,17 +19,15 @@ public final class MockControllerMetrics implements ControllerMetrics { private volatile boolean active; - private volatile int topics; - private volatile int partitions; private volatile int offlinePartitions; private volatile int preferredReplicaImbalances; + private volatile int queueSize; public MockControllerMetrics() { this.active = false; - this.topics = 0; - this.partitions = 0; this.offlinePartitions = 0; this.preferredReplicaImbalances = 0; + this.queueSize = 0; } @Override @@ -53,23 +51,13 @@ public void updateEventQueueProcessingTime(long durationMs) { } @Override - public void setGlobalTopicsCount(int topicCount) { - this.topics = topicCount; + public void setEventQueueSize(int queueSize) { + this.queueSize = queueSize; } @Override - public int globalTopicsCount() { - return this.topics; - } - - @Override - public void setGlobalPartitionCount(int partitionCount) { - this.partitions = partitionCount; - } - - @Override - public int globalPartitionCount() { - return this.partitions; + public int eventQueueSize() { + return this.queueSize; } @Override diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index badd1f86dd73e..99ed4b204d8c3 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -205,53 +205,6 @@ public void testCreateTopics() throws Exception { ctx.replicationControl.iterator(Long.MAX_VALUE)); } - @Test - public void testGlobalTopicAndPartitionMetrics() throws Exception { - ReplicationControlTestContext ctx = new ReplicationControlTestContext(); - ReplicationControlManager replicationControl = ctx.replicationControl; - CreateTopicsRequestData request = new CreateTopicsRequestData(); - request.topics().add(new CreatableTopic().setName("foo"). - setNumPartitions(1).setReplicationFactor((short) -1)); - - registerBroker(0, ctx); - unfenceBroker(0, ctx); - registerBroker(1, ctx); - unfenceBroker(1, ctx); - registerBroker(2, ctx); - unfenceBroker(2, ctx); - - List topicsToDelete = new ArrayList<>(); - - ControllerResult result = - replicationControl.createTopics(request); - topicsToDelete.add(result.response().topics().find("foo").topicId()); - - ControllerTestUtils.replayAll(replicationControl, result.records()); - assertEquals(1, ctx.metrics.globalTopicsCount()); - - request = new CreateTopicsRequestData(); - request.topics().add(new CreatableTopic().setName("bar"). - setNumPartitions(1).setReplicationFactor((short) -1)); - request.topics().add(new CreatableTopic().setName("baz"). - setNumPartitions(2).setReplicationFactor((short) -1)); - result = replicationControl.createTopics(request); - ControllerTestUtils.replayAll(replicationControl, result.records()); - assertEquals(3, ctx.metrics.globalTopicsCount()); - assertEquals(4, ctx.metrics.globalPartitionCount()); - - topicsToDelete.add(result.response().topics().find("baz").topicId()); - ControllerResult> deleteResult = replicationControl.deleteTopics(topicsToDelete); - ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); - assertEquals(1, ctx.metrics.globalTopicsCount()); - assertEquals(1, ctx.metrics.globalPartitionCount()); - - Uuid topicToDelete = result.response().topics().find("bar").topicId(); - deleteResult = replicationControl.deleteTopics(Collections.singletonList(topicToDelete)); - ControllerTestUtils.replayAll(replicationControl, deleteResult.records()); - assertEquals(0, ctx.metrics.globalTopicsCount()); - assertEquals(0, ctx.metrics.globalPartitionCount()); - } - @Test public void testOfflinePartitionAndReplicaImbalanceMetrics() throws Exception { ReplicationControlTestContext ctx = new ReplicationControlTestContext(); @@ -262,13 +215,13 @@ public void testOfflinePartitionAndReplicaImbalanceMetrics() throws Exception { unfenceBroker(i, ctx); } - ctx.createTestTopic("foo", + CreatableTopicResult foo = ctx.createTestTopic("foo", new int[][] { new int[] {0, 2}, new int[] {0, 1} }); - ctx.createTestTopic("zar", + CreatableTopicResult zar = ctx.createTestTopic("zar", new int[][] { new int[] {0, 1, 2}, new int[] {1, 2, 3}, @@ -303,6 +256,20 @@ public void testOfflinePartitionAndReplicaImbalanceMetrics() throws Exception { // After unregistering broker 3 the last partition for topic zar should go offline assertEquals(5, ctx.metrics.offlinePartitionCount()); + + // Deleting topic foo should bring the offline partition count down to 3 + ArrayList records = new ArrayList<>(); + replicationControl.deleteTopic(foo.topicId(), records); + ctx.replay(records); + + assertEquals(3, ctx.metrics.offlinePartitionCount()); + + // Deleting topic zar should bring the offline partition count down to 0 + records = new ArrayList<>(); + replicationControl.deleteTopic(zar.topicId(), records); + ctx.replay(records); + + assertEquals(0, ctx.metrics.offlinePartitionCount()); } From 210d3a75ef62389cc29806acaa8d81ffa3028092 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Thu, 13 May 2021 12:19:49 -0700 Subject: [PATCH 06/11] moving-q-size --- .../kafka/controller/ControllerMetrics.java | 4 ---- .../kafka/controller/QuorumController.java | 8 ------- .../controller/QuorumControllerMetrics.java | 21 ------------------- .../apache/kafka/queue/KafkaEventQueue.java | 14 ------------- .../controller/MockControllerMetrics.java | 12 ----------- 5 files changed, 59 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java index afb06862ca3af..d334b2d2073cd 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ControllerMetrics.java @@ -27,10 +27,6 @@ public interface ControllerMetrics { void updateEventQueueProcessingTime(long durationMs); - void setEventQueueSize(int queueSize); - - int eventQueueSize(); - void setOfflinePartitionCount(int offlinePartitions); int offlinePartitionCount(); diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 38499a865326a..42947af729310 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -331,7 +331,6 @@ public String toString() { private void appendControlEvent(String name, Runnable handler) { ControlEvent event = new ControlEvent(name, handler); queue.append(event); - controllerMetrics.setEventQueueSize(queue.size()); } private static final String GENERATE_SNAPSHOT = "generateSnapshot"; @@ -375,14 +374,12 @@ void cancel() { generator.writer().close(); generator = null; queue.cancelDeferred(GENERATE_SNAPSHOT); - controllerMetrics.setEventQueueSize(queue.size()); } void reschedule(long delayNs) { ControlEvent event = new ControlEvent(GENERATE_SNAPSHOT, this); queue.scheduleDeferred(event.name, new EarliestDeadlineFunction(time.nanoseconds() + delayNs), event); - controllerMetrics.setEventQueueSize(queue.size()); } @Override @@ -475,7 +472,6 @@ ReplicationControlManager replicationControl() { CompletableFuture appendReadEvent(String name, Supplier handler) { ControllerReadEvent event = new ControllerReadEvent(name, handler); queue.append(event); - controllerMetrics.setEventQueueSize(queue.size()); return event.future(); } @@ -611,7 +607,6 @@ private CompletableFuture appendWriteEvent(String name, ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); queue.appendWithDeadline(time.nanoseconds() + NANOSECONDS.convert(timeoutMs, TimeUnit.MILLISECONDS), event); - controllerMetrics.setEventQueueSize(queue.size()); return event.future(); } @@ -619,7 +614,6 @@ private CompletableFuture appendWriteEvent(String name, ControllerWriteOperation op) { ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); queue.append(event); - controllerMetrics.setEventQueueSize(queue.size()); return event.future(); } @@ -714,7 +708,6 @@ private void scheduleDeferredWriteEvent(String name, long deadlineNs, ControllerWriteOperation op) { ControllerWriteEvent event = new ControllerWriteEvent<>(name, op); queue.scheduleDeferred(name, new EarliestDeadlineFunction(deadlineNs), event); - controllerMetrics.setEventQueueSize(queue.size()); event.future.exceptionally(e -> { if (e instanceof UnknownServerException && e.getCause() != null && e.getCause() instanceof RejectedExecutionException) { @@ -751,7 +744,6 @@ private void rescheduleMaybeFenceStaleBrokers() { private void cancelMaybeFenceReplicas() { queue.cancelDeferred(MAYBE_FENCE_REPLICAS); - controllerMetrics.setEventQueueSize(queue.size()); } @SuppressWarnings("unchecked") diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index 3fc8975fc6697..8e0d0d20c3e08 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -30,8 +30,6 @@ public final class QuorumControllerMetrics implements ControllerMetrics { "kafka.controller", "ControllerEventManager", "EventQueueTimeMs", null); private final static MetricName EVENT_QUEUE_PROCESSING_TIME_MS = new MetricName( "kafka.controller", "ControllerEventManager", "EventQueueProcessingTimeMs", null); - private final static MetricName EVENT_QUEUE_SIZE = new MetricName( - "kafka.controller", "ControllerEventManager", "EventQueueSize", null); private final static MetricName OFFLINE_PARTITION_COUNT = new MetricName( "kafka.controller", "ReplicationControlManager", "OfflinePartitionCount", null); private final static MetricName PREFERRED_REPLICA_IMBALANCE_COUNT = new MetricName( @@ -41,11 +39,9 @@ public final class QuorumControllerMetrics implements ControllerMetrics { private volatile boolean active; private volatile int offlinePartitions; private volatile int preferredReplicaImbalances; - private volatile int queueSize; private final Gauge activeControllerCount; private final Gauge offlinePartitionCount; private final Gauge preferredReplicaImbalanceCount; - private final Gauge eventQueueSize; private final Histogram eventQueueTime; private final Histogram eventQueueProcessingTime; @@ -53,7 +49,6 @@ public QuorumControllerMetrics(MetricsRegistry registry) { this.active = false; this.offlinePartitions = 0; this.preferredReplicaImbalances = 0; - this.queueSize = 0; this.activeControllerCount = registry.newGauge(ACTIVE_CONTROLLER_COUNT, new Gauge() { @Override public Integer value() { @@ -62,12 +57,6 @@ public Integer value() { }); this.eventQueueTime = registry.newHistogram(EVENT_QUEUE_TIME_MS, true); this.eventQueueProcessingTime = registry.newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true); - this.eventQueueSize = registry.newGauge(EVENT_QUEUE_SIZE, new Gauge() { - @Override - public Integer value() { - return queueSize; - } - }); this.offlinePartitionCount = registry.newGauge(OFFLINE_PARTITION_COUNT, new Gauge() { @Override public Integer value() { @@ -102,16 +91,6 @@ public void updateEventQueueProcessingTime(long durationMs) { eventQueueTime.update(durationMs); } - @Override - public void setEventQueueSize(int queueSize) { - this.queueSize = queueSize; - } - - @Override - public int eventQueueSize() { - return this.queueSize; - } - @Override public void setOfflinePartitionCount(int offlinePartitions) { this.offlinePartitions = offlinePartitions; diff --git a/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java b/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java index 209ee19e3e363..05642f6bac8b5 100644 --- a/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java +++ b/metadata/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java @@ -157,8 +157,6 @@ private class EventHandler implements Runnable { */ private final EventContext head = new EventContext(null, null, null); - private int queueSize = 0; - /** * An ordered map of times in monotonic nanoseconds to events to time out. */ @@ -179,14 +177,8 @@ public void run() { } } - public int queueSize() { - return this.queueSize; - } - private void remove(EventContext eventContext) { eventContext.remove(); - System.out.println("removing" + eventContext.toString()); - this.queueSize--; if (eventContext.deadlineNs.isPresent()) { deadlineMap.remove(eventContext.deadlineNs.getAsLong()); eventContext.deadlineNs = OptionalLong.empty(); @@ -285,8 +277,6 @@ Exception enqueue(EventContext eventContext, OptionalLong deadlineNs = deadlineNsCalculator.apply(existingDeadlineNs); boolean queueWasEmpty = head.isSingleton(); boolean shouldSignal = false; - queueSize++; - System.out.println(eventContext.toString() + eventContext.insertionType.toString() + " " + queueSize); switch (eventContext.insertionType) { case APPEND: head.insertBefore(eventContext); @@ -380,10 +370,6 @@ public KafkaEventQueue(Time time, this.eventHandlerThread.start(); } - public int size() { - return this.eventHandler.queueSize(); - } - @Override public void enqueue(EventInsertionType insertionType, String tag, diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index 2412bd388fce6..5b373036e7370 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -21,13 +21,11 @@ public final class MockControllerMetrics implements ControllerMetrics { private volatile boolean active; private volatile int offlinePartitions; private volatile int preferredReplicaImbalances; - private volatile int queueSize; public MockControllerMetrics() { this.active = false; this.offlinePartitions = 0; this.preferredReplicaImbalances = 0; - this.queueSize = 0; } @Override @@ -50,16 +48,6 @@ public void updateEventQueueProcessingTime(long durationMs) { // nothing to do } - @Override - public void setEventQueueSize(int queueSize) { - this.queueSize = queueSize; - } - - @Override - public int eventQueueSize() { - return this.queueSize; - } - @Override public void setOfflinePartitionCount(int offlinePartitions) { this.offlinePartitions = offlinePartitions; From 1bee0ff6cb03208df77f5c46161a4019176a5fa6 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Thu, 13 May 2021 12:36:26 -0700 Subject: [PATCH 07/11] nit --- .../main/java/org/apache/kafka/controller/QuorumController.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 0b71f223fc76e..7995b4b5134ea 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -821,7 +821,7 @@ private void replay(ApiMessage message, long snapshotEpoch, long offset) { /** * The controller metrics. */ - final ControllerMetrics controllerMetrics; + private final ControllerMetrics controllerMetrics; /** * A registry for snapshot data. This must be accessed only by the event queue thread. From 5a44fcbb271c1188f7c6b834fa2ae33b1f7855cd Mon Sep 17 00:00:00 2001 From: dielhennr Date: Thu, 13 May 2021 12:38:02 -0700 Subject: [PATCH 08/11] nit --- .../org/apache/kafka/controller/QuorumControllerMetrics.java | 1 + 1 file changed, 1 insertion(+) diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java index 3e41574dad5c7..52abc8ce198d6 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumControllerMetrics.java @@ -112,6 +112,7 @@ public void updateEventQueueProcessingTime(long durationMs) { eventQueueTime.update(durationMs); } + @Override public void setGlobalTopicsCount(int topicCount) { this.globalTopicCount = topicCount; } From 070dc3f1b2d0868efa0becf6c052d3b768202030 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Mon, 17 May 2021 11:48:12 -0700 Subject: [PATCH 09/11] address-comments --- .../apache/kafka/controller/BrokersToIsrs.java | 16 +++------------- .../controller/ReplicationControlManager.java | 1 - 2 files changed, 3 insertions(+), 14 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java index 12b798d7d180b..8e54beff8c910 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java +++ b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java @@ -24,7 +24,6 @@ import java.util.Arrays; import java.util.Collections; import java.util.Iterator; -import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; import java.util.NoSuchElementException; @@ -141,14 +140,11 @@ public TopicIdPartition next() { */ private final TimelineHashMap> isrMembers; - private final Map offlinePartitionCounts; - private int offlinePartitionCount; BrokersToIsrs(SnapshotRegistry snapshotRegistry) { this.snapshotRegistry = snapshotRegistry; this.isrMembers = new TimelineHashMap<>(snapshotRegistry, 0); - this.offlinePartitionCounts = new HashMap<>(); this.offlinePartitionCount = 0; } @@ -171,8 +167,6 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, if (prevLeader == NO_LEADER) { prev = Replicas.copyWith(prevIsr, NO_LEADER); if (nextLeader != NO_LEADER) { - int offlinePartitionsForTopic = offlinePartitionCounts.getOrDefault(topicId, 0); - offlinePartitionCounts.put(topicId, --offlinePartitionsForTopic); offlinePartitionCount--; } } else { @@ -187,8 +181,6 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, if (nextLeader == NO_LEADER) { next = Replicas.copyWith(nextIsr, NO_LEADER); if (prevLeader != NO_LEADER) { - int offlinePartitionsForTopic = offlinePartitionCounts.getOrDefault(topicId, 0); - offlinePartitionCounts.put(topicId, ++offlinePartitionsForTopic); offlinePartitionCount++; } } else { @@ -234,15 +226,13 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, void removeTopicEntryForBroker(Uuid topicId, int brokerId) { Map topicMap = isrMembers.get(brokerId); if (topicMap != null) { + if (brokerId == NO_LEADER) { + offlinePartitionCount -= topicMap.get(topicId).length; + } topicMap.remove(topicId); } } - void updateMetricsForTopicRemoval(Uuid topicId) { - offlinePartitionCount = offlinePartitionCount - offlinePartitionCounts.getOrDefault(topicId, 0); - offlinePartitionCounts.put(topicId, 0); - } - private void add(int brokerId, Uuid topicId, int newPartition, boolean leader) { if (leader) { newPartition = newPartition | LEADER_FLAG; diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 3c220ccfb409f..413616439bb0b 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -427,7 +427,6 @@ public void replay(RemoveTopicRecord record) { } globalPartitionCount--; } - brokersToIsrs.updateMetricsForTopicRemoval(topic.id); brokersToIsrs.removeTopicEntryForBroker(topic.id, NO_LEADER); controllerMetrics.setGlobalTopicsCount(topics.size()); From 08d213bd7bb3bb3d56e4da07806ff855331cd0b0 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Mon, 17 May 2021 12:01:01 -0700 Subject: [PATCH 10/11] using-timeline-int --- .../apache/kafka/controller/BrokersToIsrs.java | 13 +++++++------ .../controller/ReplicationControlManager.java | 16 ++++++++-------- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java index 8e54beff8c910..42f3c4eba748d 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java +++ b/metadata/src/main/java/org/apache/kafka/controller/BrokersToIsrs.java @@ -20,6 +20,7 @@ import org.apache.kafka.common.Uuid; import org.apache.kafka.timeline.SnapshotRegistry; import org.apache.kafka.timeline.TimelineHashMap; +import org.apache.kafka.timeline.TimelineInteger; import java.util.Arrays; import java.util.Collections; @@ -140,12 +141,12 @@ public TopicIdPartition next() { */ private final TimelineHashMap> isrMembers; - private int offlinePartitionCount; + private final TimelineInteger offlinePartitionCount; BrokersToIsrs(SnapshotRegistry snapshotRegistry) { this.snapshotRegistry = snapshotRegistry; this.isrMembers = new TimelineHashMap<>(snapshotRegistry, 0); - this.offlinePartitionCount = 0; + this.offlinePartitionCount = new TimelineInteger(snapshotRegistry); } /** @@ -167,7 +168,7 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, if (prevLeader == NO_LEADER) { prev = Replicas.copyWith(prevIsr, NO_LEADER); if (nextLeader != NO_LEADER) { - offlinePartitionCount--; + offlinePartitionCount.decrement(); } } else { prev = Replicas.clone(prevIsr); @@ -181,7 +182,7 @@ void update(Uuid topicId, int partitionId, int[] prevIsr, int[] nextIsr, if (nextLeader == NO_LEADER) { next = Replicas.copyWith(nextIsr, NO_LEADER); if (prevLeader != NO_LEADER) { - offlinePartitionCount++; + offlinePartitionCount.increment(); } } else { next = Replicas.clone(nextIsr); @@ -227,7 +228,7 @@ void removeTopicEntryForBroker(Uuid topicId, int brokerId) { Map topicMap = isrMembers.get(brokerId); if (topicMap != null) { if (brokerId == NO_LEADER) { - offlinePartitionCount -= topicMap.get(topicId).length; + offlinePartitionCount.set(offlinePartitionCount.get() - topicMap.get(topicId).length); } topicMap.remove(topicId); } @@ -340,6 +341,6 @@ boolean hasLeaderships(int brokerId) { } int offlinePartitionCount() { - return offlinePartitionCount; + return offlinePartitionCount.get(); } } diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index 0723a8410f1fe..aadf50fc169f1 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -288,7 +288,7 @@ public String toString() { /** * A count of the number of partitions that do not have their first replica as a leader. */ - private int preferredReplicaImbalanceCount; + private final TimelineInteger preferredReplicaImbalanceCount; /** * A reference to the controller's configuration control manager. @@ -335,7 +335,7 @@ public String toString() { this.controllerMetrics = controllerMetrics; this.clusterControl = clusterControl; this.globalPartitionCount = new TimelineInteger(snapshotRegistry); - this.preferredReplicaImbalanceCount = 0; + this.preferredReplicaImbalanceCount = new TimelineInteger(snapshotRegistry); this.topicsByName = new TimelineHashMap<>(snapshotRegistry, 0); this.topics = new TimelineHashMap<>(snapshotRegistry, 0); this.brokersToIsrs = new BrokersToIsrs(snapshotRegistry); @@ -373,10 +373,10 @@ public void replay(PartitionRecord record) { newPartInfo.isr, prevPartInfo.leader, newPartInfo.leader); } if (newPartInfo.leader != newPartInfo.preferredReplica()) { - preferredReplicaImbalanceCount++; + preferredReplicaImbalanceCount.increment(); } controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); - controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount.get()); } public void replay(PartitionChangeRecord record) { @@ -400,10 +400,10 @@ public void replay(PartitionChangeRecord record) { newPartitionInfo.maybeLogPartitionChange(log, topicPart, prevPartitionInfo); if ((newPartitionInfo.leader != newPartitionInfo.preferredReplica()) && (prevPartitionInfo.leader == prevPartitionInfo.preferredReplica())) { - preferredReplicaImbalanceCount++; + preferredReplicaImbalanceCount.increment(); } controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); - controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount.get()); } public void replay(RemoveTopicRecord record) { @@ -424,7 +424,7 @@ public void replay(RemoveTopicRecord record) { brokersToIsrs.removeTopicEntryForBroker(topic.id, partition.isr[i]); } if (partition.leader != partition.preferredReplica()) { - preferredReplicaImbalanceCount--; + preferredReplicaImbalanceCount.decrement(); } globalPartitionCount.decrement(); } @@ -433,7 +433,7 @@ public void replay(RemoveTopicRecord record) { controllerMetrics.setGlobalTopicsCount(topics.size()); controllerMetrics.setGlobalPartitionCount(globalPartitionCount.get()); controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); - controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount); + controllerMetrics.setPreferredReplicaImbalanceCount(preferredReplicaImbalanceCount.get()); log.info("Removed topic {} with ID {}.", topic.name, record.topicId()); } From 53f290ca1f92011d2c2f367afab734d59a4c31a0 Mon Sep 17 00:00:00 2001 From: dielhennr Date: Thu, 20 May 2021 16:19:22 -0700 Subject: [PATCH 11/11] address-comments --- .../controller/ReplicationControlManager.java | 7 +++++-- .../kafka/controller/MockControllerMetrics.java | 1 + .../controller/ReplicationControlManagerTest.java | 15 ++++----------- 3 files changed, 10 insertions(+), 13 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index aadf50fc169f1..083061b2fa052 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -227,6 +227,10 @@ boolean hasLeader() { return leader != NO_LEADER; } + boolean hasPreferredLeader() { + return leader == preferredReplica(); + } + int preferredReplica() { return replicas.length == 0 ? NO_LEADER : replicas[0]; } @@ -398,8 +402,7 @@ public void replay(PartitionChangeRecord record) { String topicPart = topicInfo.name + "-" + record.partitionId() + " with topic ID " + record.topicId(); newPartitionInfo.maybeLogPartitionChange(log, topicPart, prevPartitionInfo); - if ((newPartitionInfo.leader != newPartitionInfo.preferredReplica()) && - (prevPartitionInfo.leader == prevPartitionInfo.preferredReplica())) { + if (!newPartitionInfo.hasPreferredLeader() && prevPartitionInfo.hasPreferredLeader()) { preferredReplicaImbalanceCount.increment(); } controllerMetrics.setOfflinePartitionCount(brokersToIsrs.offlinePartitionCount()); diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java index 31c7eb5575017..844475abfc719 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockControllerMetrics.java @@ -53,6 +53,7 @@ public void updateEventQueueProcessingTime(long durationMs) { // nothing to do } + @Override public void setGlobalTopicsCount(int topicCount) { this.topics = topicCount; } diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java index 36738ec47164b..5b6c2cc735b37 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java @@ -262,18 +262,11 @@ public void testOfflinePartitionAndReplicaImbalanceMetrics() throws Exception { unfenceBroker(i, ctx); } - CreatableTopicResult foo = ctx.createTestTopic("foo", - new int[][] { - new int[] {0, 2}, - new int[] {0, 1} - }); + CreatableTopicResult foo = ctx.createTestTopic("foo", new int[][] { + new int[] {0, 2}, new int[] {0, 1}}); - CreatableTopicResult zar = ctx.createTestTopic("zar", - new int[][] { - new int[] {0, 1, 2}, - new int[] {1, 2, 3}, - new int[] {1, 2, 0} - }); + CreatableTopicResult zar = ctx.createTestTopic("zar", new int[][] { + new int[] {0, 1, 2}, new int[] {1, 2, 3}, new int[] {1, 2, 0}}); ControllerResult result = replicationControl.unregisterBroker(0); ctx.replay(result.records());