Skip to content

[fix][broker] new load balancer system topic should not be auto-created now #20566

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

Merged
Technoboy- merged 1 commit into from
Jun 14, 2023
Merged

[fix][broker] new load balancer system topic should not be auto-created now #20566

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
12 changes: 11 additions & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,7 @@
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.intercept.ManagedLedgerInterceptorImpl;
import org.apache.pulsar.broker.loadbalance.LoadManager;
import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.resources.DynamicConfigurationResources;
import org.apache.pulsar.broker.resources.LocalPoliciesResources;
Expand Down Expand Up @@ -3304,10 +3305,19 @@ private CompletableFuture<Boolean> isAllowAutoTopicCreationAsync(final TopicName
topicName.getNamespaceObject());
return CompletableFuture.completedFuture(false);
}
//System topic can always be created automatically

// ServiceUnitStateChannelImpl.TOPIC expects to be a non-partitioned-topic now.
// We don't allow the auto-creation here.
// ServiceUnitStateChannelImpl.start() is responsible to create the topic.
if (ServiceUnitStateChannelImpl.TOPIC.equals(topicName.toString())) {

@Demogorgon314 Demogorgon314 Jun 13, 2023

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.

This change makes sense to me, but we should finally change it to a partitioned topic.

@heesung-sohn heesung-sohn Jun 13, 2023

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.

Yes this is another optimization area for scalability.

@michaeljmarshall michaeljmarshall Jun 20, 2023

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.

Why is it that in some cases we auto create in a non-partitioned way and in this case we do not auto create it?

@heesung-sohn heesung-sohn Jun 20, 2023

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.

I am not sure if I understood your question. Since the system topic is currently expected to work on a non-partitioned topic, we specifically create the topic in ServiceUnitStateChannelImpl.start(). This change will block the system topic auto-creation(when any other broker tries to auto-create the topic due to any race-condition )

@michaeljmarshall michaeljmarshall Jun 20, 2023

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.

Sorry, I was mixing up #20370 and #20397.

My primary concern is that the solution in this PR creates a special case instead of creating a framework to make it easier to build system topics in the future. In my opinion, the load manager should be contained by the relevant interfaces and should not be referenced here. There is some relevant discussion about system topics in this PR too #20514 (comment). There is also relevant discussion on the ML here https://lists.apache.org/thread/f0q8n0hf1lgw9r2j53tm4yjjfdyr9kjd.

I think it is relevant that until now, we have always allowed system topics to be auto created to prevent classes of failures.

If I were to guess, the problem this PR is trying to solve is actually more generic than the system topics we're working with here. It is likely a challenge for all pulsar users to know how to create their topics correctly. That is one reason I want to push for thinking about a general solution.

@heesung-sohn heesung-sohn Jun 20, 2023

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.

I think we can make this fix general like the followings.

  1. Maintain a White-list for system topics that are pre-created. We can check this list and reject any auto creation here.

  2. Or Make Pulsar pre-create all system topics, and then we can reject auto-creation for all topics under the system namespace.

I think the first option is less intrusive.

@michaeljmarshall michaeljmarshall Jun 28, 2023

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.

A third option is to only auto create topics as non-partitioned topics. Any component that needs to create a system topic as a partitioned topic could do so on start up.

@michaeljmarshall michaeljmarshall Jun 28, 2023

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 proposed as much here on the ML: https://lists.apache.org/thread/1lndgf1hx821fc12t0pc6j6zdrhkntht.

return CompletableFuture.completedFuture(false);
}

//Other system topics can be created automatically
if (pulsar.getConfiguration().isSystemTopicEnabled() && isSystemTopic(topicName)) {
return CompletableFuture.completedFuture(true);
}

final boolean allowed;
AutoTopicCreationOverride autoTopicCreationOverride = getAutoTopicCreationOverride(topicName, policies);
if (autoTopicCreationOverride != null) {
Expand Down
11 changes: 10 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
Expand Down Expand Up @@ -1579,7 +1580,6 @@ public void testDynamicConfigurationsForceDeleteTenantAllowed() throws Exception
});
}


// this test is disabled since it is flaky
@Test(enabled = false)
public void testBrokerStatsTopicLoadFailed() throws Exception {
Expand Down Expand Up @@ -1657,4 +1657,13 @@ public void testBrokerStatsTopicLoadFailed() throws Exception {
return flag.get();
});
}

@Test
public void testIsSystemTopicAllowAutoTopicCreationAsync() throws Exception {
BrokerService brokerService = pulsar.getBrokerService();
assertFalse(brokerService.isAllowAutoTopicCreationAsync(
ServiceUnitStateChannelImpl.TOPIC).get());
assertTrue(brokerService.isAllowAutoTopicCreationAsync(
"persistent://pulsar/system/my-system-topic").get());
}
}