You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2026/04/03 16:17:51 UTC
[PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
poorbarcode opened a new pull request, #25461:
URL: https://github.com/apache/pulsar/pull/25461
### Motivation
This is part-3 of PIP-433, see the goal 1 in [PIP-433](https://github.com/apache/pulsar/blob/master/pip/pip-433.md)
### Modifications
always allow the replicator to register a new compatible schema, users can turn off this function
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3131578186
##########
pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java:
##########
@@ -1361,10 +1361,12 @@ public static ByteBuf newGetSchemaResponseError(long requestId, ServerError erro
return serializeWithSize(newGetSchemaResponseErrorCommand(requestId, error, errorMessage));
}
- public static ByteBuf newGetOrCreateSchema(long requestId, String topic, SchemaInfo schemaInfo) {
+ public static ByteBuf newGetOrCreateSchema(long requestId, String topic, String producerName,
+ SchemaInfo schemaInfo) {
BaseCommand cmd = localCmd(Type.GET_OR_CREATE_SCHEMA);
Schema schema = cmd.setGetOrCreateSchema()
.setRequestId(requestId)
+ .setProducerName(producerName)
Review Comment:
I will also add a test for the case then
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3131572593
##########
pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java:
##########
@@ -1361,10 +1361,12 @@ public static ByteBuf newGetSchemaResponseError(long requestId, ServerError erro
return serializeWithSize(newGetSchemaResponseErrorCommand(requestId, error, errorMessage));
}
- public static ByteBuf newGetOrCreateSchema(long requestId, String topic, SchemaInfo schemaInfo) {
+ public static ByteBuf newGetOrCreateSchema(long requestId, String topic, String producerName,
+ SchemaInfo schemaInfo) {
BaseCommand cmd = localCmd(Type.GET_OR_CREATE_SCHEMA);
Schema schema = cmd.setGetOrCreateSchema()
.setRequestId(requestId)
+ .setProducerName(producerName)
Review Comment:
@Denovo1998
> Seems like there are some issues? https://github.com/apache/pulsar/pull/25461#issuecomment-4296557751
Yes, I forgot to modify the client-side code. I added the change now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3107901475
##########
pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java:
##########
@@ -218,6 +222,8 @@ public boolean equals(Object obj) {
&& schema_validation_enforced == other.schema_validation_enforced
&& schema_compatibility_strategy == other.schema_compatibility_strategy
&& is_allow_auto_update_schema == other.is_allow_auto_update_schema
+ && is_allow_auto_update_schema_with_replicator
+ == other.is_allow_auto_update_schema_with_replicator
Review Comment:
Good suggestion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3107930061
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java:
##########
@@ -2163,14 +2163,20 @@ private class SetIsAllowAutoUpdateSchema extends CliCommand {
@Option(names = { "--disable", "-d" }, description = "Disable schema validation enforced")
private boolean disable = false;
+ @Option(names = { "--disable-for-replicator" },
+ description = "By default, brokers always allow replicator to register new compatible schemas even"
+ + " when auto updates are disabled, if you want to disable replicators to register compatible schemas,"
+ + " please set it to true")
+ private boolean disableForReplicator;
Review Comment:
Fixed
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java:
##########
@@ -2163,14 +2163,20 @@ private class SetIsAllowAutoUpdateSchema extends CliCommand {
@Option(names = { "--disable", "-d" }, description = "Disable schema validation enforced")
private boolean disable = false;
+ @Option(names = { "--disable-for-replicator" },
+ description = "By default, brokers always allow replicator to register new compatible schemas even"
+ + " when auto updates are disabled, if you want to disable replicators to register compatible schemas,"
+ + " please set it to true")
+ private boolean disableForReplicator;
+
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3107885954
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -2748,7 +2750,13 @@ protected void handleGetOrCreateSchema(CommandGetOrCreateSchema commandGetOrCrea
service.getTopicIfExists(topicName).thenAccept(topicOpt -> {
if (topicOpt.isPresent()) {
Topic topic = topicOpt.get();
- CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);
+ boolean isReplicatorProducer = false;
+ if (commandGetOrCreateSchema.hasProducerName()) {
+ isReplicatorProducer = Producer.isRemoteOrShadow(commandGetOrCreateSchema.getProducerName(),
+ getBrokerService().getPulsar().getConfig().getReplicatorPrefix());
+ }
+ CompletableFuture<SchemaVersion> schemaVersionFuture =
+ tryAddSchema(topic, schema, isReplicatorProducer);
Review Comment:
Yes, it will.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3089854693
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java:
##########
@@ -2748,7 +2750,13 @@ protected void handleGetOrCreateSchema(CommandGetOrCreateSchema commandGetOrCrea
service.getTopicIfExists(topicName).thenAccept(topicOpt -> {
if (topicOpt.isPresent()) {
Topic topic = topicOpt.get();
- CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);
+ boolean isReplicatorProducer = false;
+ if (commandGetOrCreateSchema.hasProducerName()) {
+ isReplicatorProducer = Producer.isRemoteOrShadow(commandGetOrCreateSchema.getProducerName(),
+ getBrokerService().getPulsar().getConfig().getReplicatorPrefix());
+ }
+ CompletableFuture<SchemaVersion> schemaVersionFuture =
+ tryAddSchema(topic, schema, isReplicatorProducer);
Review Comment:
Will it be a security issue that user can just set producer name to pulsar.repl.* to get rid of the schema upload control? if user disabled is_allow_auto_update_schema
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java:
##########
@@ -2163,14 +2163,20 @@ private class SetIsAllowAutoUpdateSchema extends CliCommand {
@Option(names = { "--disable", "-d" }, description = "Disable schema validation enforced")
private boolean disable = false;
+ @Option(names = { "--disable-for-replicator" },
+ description = "By default, brokers always allow replicator to register new compatible schemas even"
+ + " when auto updates are disabled, if you want to disable replicators to register compatible schemas,"
+ + " please set it to true")
+ private boolean disableForReplicator;
+
Review Comment:
--disable-for-replicator is a primitive boolean defaulting to false, so every set-is-allow-auto-update-schema call sends allowAutoUpdateSchemaWithReplicator=true unless the flag is supplied. A user who ran ... --disable --disable-for-replicator and later runs ... --enable silently re-enables replicator updates — different from the REST API's null-preserving semantics.
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java:
##########
@@ -2163,14 +2163,20 @@ private class SetIsAllowAutoUpdateSchema extends CliCommand {
@Option(names = { "--disable", "-d" }, description = "Disable schema validation enforced")
private boolean disable = false;
+ @Option(names = { "--disable-for-replicator" },
+ description = "By default, brokers always allow replicator to register new compatible schemas even"
+ + " when auto updates are disabled, if you want to disable replicators to register compatible schemas,"
+ + " please set it to true")
+ private boolean disableForReplicator;
Review Comment:
PIP says CLI flag is --enable-for-replicator, implementation uses --disable-for-replicator.
##########
pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/Policies.java:
##########
@@ -218,6 +222,8 @@ public boolean equals(Object obj) {
&& schema_validation_enforced == other.schema_validation_enforced
&& schema_compatibility_strategy == other.schema_compatibility_strategy
&& is_allow_auto_update_schema == other.is_allow_auto_update_schema
+ && is_allow_auto_update_schema_with_replicator
+ == other.is_allow_auto_update_schema_with_replicator
Review Comment:
Should use Objects.equals()?
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java:
##########
@@ -2163,14 +2163,20 @@ private class SetIsAllowAutoUpdateSchema extends CliCommand {
@Option(names = { "--disable", "-d" }, description = "Disable schema validation enforced")
private boolean disable = false;
+ @Option(names = { "--disable-for-replicator" },
+ description = "By default, brokers always allow replicator to register new compatible schemas even"
+ + " when auto updates are disabled, if you want to disable replicators to register compatible schemas,"
+ + " please set it to true")
+ private boolean disableForReplicator;
+
Review Comment:
Please also check the Java API
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -2571,13 +2571,18 @@ protected void internalSetSchemaValidationEnforced(boolean schemaValidationEnfor
"schemaValidationEnforced");
}
- protected void internalSetIsAllowAutoUpdateSchema(boolean isAllowAutoUpdateSchema) {
+ protected void internalSetIsAllowAutoUpdateSchema(boolean isAllowAutoUpdateSchema,
+ Boolean isAllowAutoUpdateSchemaWithReplicator) {
validateNamespacePolicyOperation(namespaceName, PolicyName.SCHEMA_COMPATIBILITY_STRATEGY,
PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
mutatePolicy((policies) -> {
policies.is_allow_auto_update_schema = isAllowAutoUpdateSchema;
+ if (isAllowAutoUpdateSchemaWithReplicator != null) {
+ policies.is_allow_auto_update_schema_with_replicator =
+ isAllowAutoUpdateSchemaWithReplicator;
+ }
return policies;
}, (policies) -> policies.is_allow_auto_update_schema,
"isAllowAutoUpdateSchema");
Review Comment:
Missed isAllowAutoUpdateSchemaWithReplicator?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode merged PR #25461:
URL: https://github.com/apache/pulsar/pull/25461
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#issuecomment-4256278167
@claude review this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3107898718
##########
pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java:
##########
@@ -2163,14 +2163,20 @@ private class SetIsAllowAutoUpdateSchema extends CliCommand {
@Option(names = { "--disable", "-d" }, description = "Disable schema validation enforced")
private boolean disable = false;
+ @Option(names = { "--disable-for-replicator" },
+ description = "By default, brokers always allow replicator to register new compatible schemas even"
+ + " when auto updates are disabled, if you want to disable replicators to register compatible schemas,"
+ + " please set it to true")
+ private boolean disableForReplicator;
+
Review Comment:
> --disable-for-replicator is a primitive boolean defaulting to false, so every set-is-allow-auto-update-schema call sends allowAutoUpdateSchemaWithReplicator=true unless the flag is supplied. A user who ran ... --disable --disable-for-replicator and later runs ... --enable silently re-enables replicator updates — different from the REST API's null-preserving semantics.
This enhances the convenience of using `pulsar-admin`.
- The design of our API is such that if the boolean type needs to be set to true, the parameter to be given is' --enableXxx '.
- If 'null' needs to be supported, the user needs to input '--enabledXxx true/false'
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#discussion_r3107928646
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java:
##########
@@ -2571,13 +2571,18 @@ protected void internalSetSchemaValidationEnforced(boolean schemaValidationEnfor
"schemaValidationEnforced");
}
- protected void internalSetIsAllowAutoUpdateSchema(boolean isAllowAutoUpdateSchema) {
+ protected void internalSetIsAllowAutoUpdateSchema(boolean isAllowAutoUpdateSchema,
+ Boolean isAllowAutoUpdateSchemaWithReplicator) {
validateNamespacePolicyOperation(namespaceName, PolicyName.SCHEMA_COMPATIBILITY_STRATEGY,
PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
mutatePolicy((policies) -> {
policies.is_allow_auto_update_schema = isAllowAutoUpdateSchema;
+ if (isAllowAutoUpdateSchemaWithReplicator != null) {
+ policies.is_allow_auto_update_schema_with_replicator =
+ isAllowAutoUpdateSchemaWithReplicator;
+ }
return policies;
}, (policies) -> policies.is_allow_auto_update_schema,
"isAllowAutoUpdateSchema");
Review Comment:
It has been modified at the line-2583
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on PR #25461:
URL: https://github.com/apache/pulsar/pull/25461#issuecomment-4296557751
@poorbarcode @codelipenghui
Seems like there are some issues?
When geo-replicator creates a remote producer in the broker, it indeed carries the replicator's producer name and uses Schema.AUTO_PRODUCE_BYTES().
But the key point of AUTO_PRODUCE_BYTES is: when establishing a producer connection, it doesn't send the "specific schema of this message" all at once. The actual specific schema registration happens later during the message sending process, when encountering a new schema, it then triggers GET_OR_CREATE_SCHEMA. This call chain looks like this locally:
1. When replicator acts as a producer to establish a connection, the `ProducerImpl.connectionOpened(...)` method sends the PRODUCER command, which includes the producerName. But later, when it actually needs to register a specific schema, it follows the path `tryRegisterSchema(...) -> getOrCreateSchemaAsync(...)`.
2. The method finally calls Commands' newGetOrCreateSchema(requestId, topic, schemaInfo).
3. This command builder only wrote requestId/topic/schema, without writing producerName.
4. So the CommandGetOrCreateSchema sent from the sender to the broker always has the field producerName absent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org