Skip to content

[fix][broker] Deprecated aggregatePublisherStatsByProducerName config #18254

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
heesung-sohn wants to merge 1 commit into from
Closed

[fix][broker] Deprecated aggregatePublisherStatsByProducerName config #18254

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions conf/broker.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1461,9 +1461,8 @@ exposePreciseBacklogInPrometheus=false

splitTopicAndPartitionLabelInPrometheus=false

# If true and the client supports partial producer, aggregate publisher stats of PartitionedTopicStats by producerName.
# Otherwise, aggregate it by list index.
aggregatePublisherStatsByProducerName=false
# Deprecated: it aggregates publisher stats by producerName
aggregatePublisherStatsByProducerName=

# Interval between checks to see if cluster is migrated and marks topic migrated
# if cluster is marked migrated. Disable with value 0. (Default disabled).
Expand Down
6 changes: 3 additions & 3 deletions conf/standalone.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -987,9 +987,9 @@ exposePreciseBacklogInPrometheus=false

splitTopicAndPartitionLabelInPrometheus=false

# If true, aggregate publisher stats of PartitionedTopicStats by producerName.
# Otherwise, aggregate it by list index.
aggregatePublisherStatsByProducerName=false

# Deprecated: it aggregates publisher stats by producerName
aggregatePublisherStatsByProducerName=

### --- Schema storage --- ###
# The schema storage implementation used by this broker.
Expand Down
5 changes: 0 additions & 5 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2705,11 +2705,6 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
doc = "Stats update initial delay in seconds"
)
private int statsUpdateInitialDelayInSecs = 60;
@FieldContext(
category = CATEGORY_METRICS,
doc = "If true, aggregate publisher stats of PartitionedTopicStats by producerName"
)
private boolean aggregatePublisherStatsByProducerName = false;

/**** --- Ledger Offloading. --- ****/
/****
Expand Down
12 changes: 1 addition & 11 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,7 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
boolean isEncrypted, Map<String, String> metadata, SchemaVersion schemaVersion, long epoch,
boolean userProvidedProducerName,
ProducerAccessMode accessMode,
Optional<Long> topicEpoch,
boolean supportsPartialProducer) {
Optional<Long> topicEpoch) {
final ServiceConfiguration serviceConf = cnx.getBrokerService().pulsar().getConfiguration();

this.topic = topic;
Expand Down Expand Up @@ -129,15 +128,6 @@ public Producer(Topic topic, TransportCnx cnx, long producerId, String producerN
stats.setClientVersion(cnx.getClientVersion());
stats.setProducerName(producerName);
stats.producerId = producerId;
if (serviceConf.isAggregatePublisherStatsByProducerName() && stats.getProducerName() != null) {
// If true and the client supports partial producer,
// aggregate publisher stats of PartitionedTopicStats by producerName.
// Otherwise, aggregate it by list index.
stats.setSupportsPartialProducer(supportsPartialProducer);
} else {
// aggregate publisher stats of PartitionedTopicStats by list index.
stats.setSupportsPartialProducer(false);
}
stats.metadata = this.metadata;
stats.accessMode = Commands.convertProducerAccessMode(accessMode);

Expand Down
11 changes: 3 additions & 8 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1236,7 +1236,6 @@ protected void handleProducer(final CommandProducer cmdProducer) {
final boolean isTxnEnabled = cmdProducer.isTxnEnabled();
final String initialSubscriptionName =
cmdProducer.hasInitialSubscriptionName() ? cmdProducer.getInitialSubscriptionName() : null;
final boolean supportsPartialProducer = supportsPartialProducer();

TopicName topicName = validateTopicName(cmdProducer.getTopic(), requestId, cmdProducer);
if (topicName == null) {
Expand Down Expand Up @@ -1392,7 +1391,7 @@ protected void handleProducer(final CommandProducer cmdProducer) {

buildProducerAndAddTopic(topic, producerId, producerName, requestId, isEncrypted,
metadata, schemaVersion, epoch, userProvidedProducerName, topicName,
producerAccessMode, topicEpoch, supportsPartialProducer, producerFuture);
producerAccessMode, topicEpoch, producerFuture);
});
}).exceptionally(exception -> {
Throwable cause = exception.getCause();
Expand Down Expand Up @@ -1464,12 +1463,12 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ
boolean isEncrypted, Map<String, String> metadata, SchemaVersion schemaVersion, long epoch,
boolean userProvidedProducerName, TopicName topicName,
ProducerAccessMode producerAccessMode,
Optional<Long> topicEpoch, boolean supportsPartialProducer,
Optional<Long> topicEpoch,
CompletableFuture<Producer> producerFuture){
CompletableFuture<Void> producerQueuedFuture = new CompletableFuture<>();
Producer producer = new Producer(topic, ServerCnx.this, producerId, producerName,
getPrincipal(), isEncrypted, metadata, schemaVersion, epoch,
userProvidedProducerName, producerAccessMode, topicEpoch, supportsPartialProducer);
userProvidedProducerName, producerAccessMode, topicEpoch);

topic.addProducer(producer, producerQueuedFuture).thenAccept(newTopicEpoch -> {
if (isActive()) {
Expand Down Expand Up @@ -3016,10 +3015,6 @@ boolean supportBrokerMetadata() {
return features != null && features.isSupportsBrokerEntryMetadata();
}

boolean supportsPartialProducer() {
return features != null && features.isSupportsPartialProducer();
}

@Override
public String getClientVersion() {
return clientVersion;
Expand Down
4 changes: 0 additions & 4 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2696,7 +2696,6 @@ public void testPartitionedStatsAggregationByProducerName(String topicType) thro
cleanup();
setup();

conf.setAggregatePublisherStatsByProducerName(true);
final String topic = topicType + "://prop-xyz/ns1/test-partitioned-stats-aggregation-by-producer-name";
admin.topics().createPartitionedTopic(topic, 10);

Expand Down Expand Up @@ -2746,12 +2745,10 @@ public int choosePartition(Message<?> msg, TopicMetadata metadata) {
assertEquals(topicStats.getPartitions().values().stream().mapToInt(e -> e.getPublishers().size()).sum(), 10);
assertEquals(topicStats.getPartitions().values().stream().map(e -> e.getPublishers().get(0).getProducerName()).distinct().count(), 2);
assertEquals(topicStats.getPublishers().size(), 2);
topicStats.getPublishers().forEach(p -> assertTrue(p.isSupportsPartialProducer()));
}

@Test(dataProvider = "topicType")
public void testPartitionedStatsAggregationByProducerNamePerPartition(String topicType) throws Exception {
conf.setAggregatePublisherStatsByProducerName(true);
final String topic = topicType + "://prop-xyz/ns1/test-partitioned-stats-aggregation-by-producer-name-per-pt";
admin.topics().createPartitionedTopic(topic, 2);

Expand All @@ -2770,7 +2767,6 @@ public void testPartitionedStatsAggregationByProducerNamePerPartition(String top
assertEquals(topicStats.getPartitions().values().stream().mapToInt(e -> e.getPublishers().size()).sum(), 2);
assertEquals(topicStats.getPartitions().values().stream().map(e -> e.getPublishers().get(0).getProducerName()).distinct().count(), 2);
assertEquals(topicStats.getPublishers().size(), 2);
topicStats.getPublishers().forEach(p -> assertTrue(p.isSupportsPartialProducer()));
}

@Test(dataProvider = "topicType")
Expand Down
34 changes: 17 additions & 17 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -438,7 +438,7 @@ public void testAddRemoveProducer() throws Exception {
// 1. simple add producer
Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, false,
ProducerAccessMode.Shared, Optional.empty(), true);
ProducerAccessMode.Shared, Optional.empty());
topic.addProducer(producer, new CompletableFuture<>());
assertEquals(topic.getProducers().size(), 1);

Expand All @@ -455,7 +455,7 @@ public void testAddRemoveProducer() throws Exception {
PersistentTopic failTopic = new PersistentTopic(failTopicName, ledgerMock, brokerService);
Producer failProducer = new Producer(failTopic, serverCnx, 2 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, false,
ProducerAccessMode.Shared, Optional.empty(), true);
ProducerAccessMode.Shared, Optional.empty());
try {
topic.addProducer(failProducer, new CompletableFuture<>());
fail("should have failed");
Expand All @@ -466,7 +466,7 @@ public void testAddRemoveProducer() throws Exception {
// 4. Try to remove with unequal producer
Producer producerCopy = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, false,
ProducerAccessMode.Shared, Optional.empty(), true);
ProducerAccessMode.Shared, Optional.empty());
topic.removeProducer(producerCopy);
// Expect producer to be in map
assertEquals(topic.getProducers().size(), 1);
Expand All @@ -485,9 +485,9 @@ public void testProducerOverwrite() throws Exception {
PersistentTopic topic = new PersistentTopic(successTopicName, ledgerMock, brokerService);
String role = "appid1";
Producer producer1 = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty());
Producer producer2 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 0, true, ProducerAccessMode.Shared, Optional.empty());
try {
topic.addProducer(producer1, new CompletableFuture<>()).join();
topic.addProducer(producer2, new CompletableFuture<>()).join();
Expand All @@ -500,7 +500,7 @@ public void testProducerOverwrite() throws Exception {
Assert.assertEquals(topic.getProducers().size(), 1);

Producer producer3 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty());

try {
topic.addProducer(producer3, new CompletableFuture<>()).join();
Expand All @@ -516,7 +516,7 @@ public void testProducerOverwrite() throws Exception {
Assert.assertEquals(topic.getProducers().size(), 0);

Producer producer4 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty());

topic.addProducer(producer3, new CompletableFuture<>());
topic.addProducer(producer4, new CompletableFuture<>());
Expand All @@ -529,21 +529,21 @@ public void testProducerOverwrite() throws Exception {
Assert.assertEquals(topic.getProducers().size(), 0);

Producer producer5 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1",
role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 1, false, ProducerAccessMode.Shared, Optional.empty());

topic.addProducer(producer5, new CompletableFuture<>());
Assert.assertEquals(topic.getProducers().size(), 1);

Producer producer6 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1",
role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 2, false, ProducerAccessMode.Shared, Optional.empty());

topic.addProducer(producer6, new CompletableFuture<>());
Assert.assertEquals(topic.getProducers().size(), 1);

topic.getProducers().values().forEach(producer -> Assert.assertEquals(producer.getEpoch(), 2));

Producer producer7 = new Producer(topic, serverCnx, 2 /* producer id */, "pulsar.repl.cluster1",
role, false, null, SchemaVersion.Latest, 3, true, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 3, true, ProducerAccessMode.Shared, Optional.empty());

topic.addProducer(producer7, new CompletableFuture<>());
Assert.assertEquals(topic.getProducers().size(), 1);
Expand All @@ -556,20 +556,20 @@ private void testMaxProducers() throws Exception {
String role = "appid1";
// 1. add producer1
Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name1", role,
false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
topic.addProducer(producer, new CompletableFuture<>());
assertEquals(topic.getProducers().size(), 1);

// 2. add producer2
Producer producer2 = new Producer(topic, serverCnx, 2 /* producer id */, "prod-name2", role,
false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
topic.addProducer(producer2, new CompletableFuture<>());
assertEquals(topic.getProducers().size(), 2);

// 3. add producer3 but reached maxProducersPerTopic
try {
Producer producer3 = new Producer(topic, serverCnx, 3 /* producer id */, "prod-name3", role,
false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
topic.addProducer(producer3, new CompletableFuture<>()).join();
fail("should have failed");
} catch (Exception e) {
Expand Down Expand Up @@ -619,7 +619,7 @@ private Producer getMockedProducerWithSpecificAddress(Topic topic, long producer
doReturn(new PulsarCommandSenderImpl(null, cnx)).when(cnx).getCommandSender();

return new Producer(topic, cnx, producerId, producerNameBase + producerId, role, false, null,
SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
}

@Test
Expand Down Expand Up @@ -1292,7 +1292,7 @@ public void testDeleteTopic() throws Exception {
// 3. delete topic with producer
topic = (PersistentTopic) brokerService.getOrCreateTopic(successTopicName).get();
Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
topic.addProducer(producer, new CompletableFuture<>()).join();

assertTrue(topic.delete().isCompletedExceptionally());
Expand Down Expand Up @@ -1466,7 +1466,7 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
String role = "appid1";
Thread.sleep(10); /* delay to ensure that the delete gets executed first */
Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty(), true);
role, false, null, SchemaVersion.Latest, 0, false, ProducerAccessMode.Shared, Optional.empty());
topic.addProducer(producer, new CompletableFuture<>()).join();
fail("Should have failed");
} catch (Exception e) {
Expand Down Expand Up @@ -2253,7 +2253,7 @@ public void testDisconnectProducer() throws Exception {
String role = "appid1";
Producer producer = new Producer(topic, serverCnx, 1 /* producer id */, "prod-name",
role, false, null, SchemaVersion.Latest, 0, false,
ProducerAccessMode.Shared, Optional.empty(), true);
ProducerAccessMode.Shared, Optional.empty());
assertFalse(producer.isDisconnecting());
// Disconnect the producer multiple times.
producer.disconnect();
Expand Down
1 change: 1 addition & 0 deletions ...client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/PublisherStats.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ public interface PublisherStats {
long getProducerId();

/** Whether partial producer is supported at client. */
@Deprecated
boolean isSupportsPartialProducer();

@Technoboy- Technoboy- Nov 1, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not remove this ?

@heesung-sohn heesung-sohn Nov 1, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is already exposed to the client interface. Are we allowed to make a breaking change?

@michaeljmarshall michaeljmarshall Nov 8, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is the correct change. If it has already been released, we cannot remove it without first deprecating it. I don't know that we have documented guidance on how long to keep methods like this marked as deprecated.


/** Producer name. */
Expand Down
Loading