You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "gaoran10 (via GitHub)" <gi...@apache.org> on 2025/07/06 17:43:10 UTC
[PR] PIP-420 Supports users implement external schemas [pulsar]
gaoran10 opened a new pull request, #24488:
URL: https://github.com/apache/pulsar/pull/24488
### Motivation
This is the implementation for [PIP-420](https://github.com/apache/pulsar/pull/24328).
### Modifications
<!-- Describe the modifications you've done. -->
### Verifying this change
Add a test to verify the Schema interface changes.
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [x] The public API
- [x] The schema
- [x] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [x] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2263048556
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if (strategy == SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE) {
+ return;
+ }
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
+ }
+
+ @Override
+ public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if (strategy == SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE) {
+ return;
+ }
+ while (from.iterator().hasNext()) {
+ SchemaData fromSchema = from.iterator().next();
+ checkCompatible(fromSchema, to, strategy);
+ }
Review Comment:
Sorry, it is here.
```java
for (SchemaData fromSchema : from) {
checkCompatible(fromSchema, to, strategy);
}
```
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299687908
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -501,16 +518,16 @@ private T decode(byte[] schemaVersion) {
}
}
- private T decodeBySchema(byte[] schemaVersion) {
- T value = poolMessage ? schema.decode(payload.nioBuffer(), schemaVersion) : null;
+ private T decodeBySchema(byte[] schemaFlag) {
Review Comment:
Changed get schemaInfo to use the schema version, changed the parameter name to `schemaIdOrVersion`.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297382704
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
Review Comment:
Does here forget to call `checkCompatible(fromSchema, to, strategy);`
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 closed pull request #24488: [improve][client] PIP-420: Supports users implement external schemas
URL: https://github.com/apache/pulsar/pull/24488
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299590642
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
Review Comment:
The external schema is not compatible with other Pulsar schemas; the external schema compatibility check will be managed by the external schema manager.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297479446
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -424,6 +424,23 @@ public byte[] getSchemaVersion() {
}
}
+ @Override
+ public byte[] getSchemaId() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
+ return null;
+ }
+
+ private byte[] getSchemaFlag() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
Review Comment:
Now if the message has schema id, the schema id will be passed to all places that need a schema version. This should be a wrong behavior because these components only retrieve the schema from **built-in** schema registry. Passing the schema id from external schema registry makes no sense.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299625349
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/EncodeData.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@AllArgsConstructor
+@Data
+public class EncodeData {
+
+ private byte[] data;
+
+ private byte[] schemaId;
+
+ public EncodeData(byte[] data) {
+ this.data = data;
+ this.schemaId = new byte[0];
+ }
Review Comment:
changed to `record`, and removed all `new EncodeData(data, null)`
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306273041
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheckTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import static org.testng.Assert.fail;
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ExternalSchemaCompatibilityCheckTest {
+
+ private final ExternalSchemaCompatibilityCheck compatibilityCheck = new ExternalSchemaCompatibilityCheck();
+
+ private final SchemaData externalSchemaData = SchemaData.builder()
+ .type(SchemaType.EXTERNAL)
+ .data(new byte[0])
+ .build();
+
+ @DataProvider(name = "otherSchemasProvider")
+ public Object[] otherSchemasProvider() {
+ return new Object[] {
+ SchemaData.builder()
+ .type(SchemaType.JSON)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.AVRO)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF_NATIVE)
+ .build()
+ };
+ }
+
+ @Test(dataProvider = "otherSchemasProvider")
+ public void testExternalSchemaCompatibilityCheck(SchemaData schemaData) {
Review Comment:
I mean this test only tests the case that `from` is a built-in schema and `to` is an external schema. It's better to test the case that `from` is an external schema and `to` is a built-in schema.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299590642
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
Review Comment:
In any strategy mode the external schema is not compatible with other Pulsar schemas; the external schema manager will manage the external schema compatibility check.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299683649
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java:
##########
@@ -208,6 +208,15 @@ public interface Message<T> {
*/
byte[] getSchemaVersion();
+ /**
+ * Get schema ID of the message.
+ * PIP-420 provides a way to produce messages with external schema,
+ * and the schema ID will be set to the message metadata.
+ *
+ * @return schema ID of the message if the message is produced with external schema.
Review Comment:
I changed to use the `Optional` object as the return type.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306279554
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java:
##########
@@ -717,4 +717,5 @@ ClientBuilder authentication(String authPluginClassName, Map<String, String> aut
* - The `loadManagerClassName` config in broker is a class that implements the `ExtensibleLoadManager` interface
*/
ClientBuilder lookupProperties(Map<String, String> properties);
+
Review Comment:
+1
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306309651
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,28 +101,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (keySchemaId == null && valueSchemaId == null) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ byteBuffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
return byteBuffer.array();
}
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
Review Comment:
Please add unit tests for this function. The current implementation is wrong.
```java
final var schemaId = generateKVSchemaId(null, new byte[]{0x11, 0x22});
final var keySchemaId = getSchemaId(schemaId, true);
```
In this case, `keySchemaId` should be an empty schema id array. However, it's the value schema.
BTW, I found allowing a null schema id could make key value schema very confusing. With the same code example, it's impossible to distinguish if the key schema is null or an empty array.
One solution is that never allow a null schema id. Then all APIs will become much simpler, e.g. `getSchemaId` won't need to return an `Optional`.
The other solution is to have special encoding for null schema id, e.g. use a negative length to represent null.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306313073
##########
pulsar-common/src/main/proto/PulsarApi.proto:
##########
@@ -173,6 +174,7 @@ message MessageMetadata {
// the `compacted_out` field in `SingleMessageMetadata` must be checked to identify and filter out compacted
// messages (e.g., `k1 => v1` and `k1 => null` in the example above).
repeated int32 compacted_batch_indexes = 31;
+ optional bytes schema_id = 32;
Review Comment:
+1
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306787045
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaImpl.java:
##########
@@ -187,8 +223,12 @@ public KeyValue<K, V> decode(byte[] keyBytes, byte[] valueBytes, byte[] schemaVe
if (valueBytes == null) {
v = null;
} else {
- if (valueSchema.supportSchemaVersioning() && schemaVersion != null) {
- v = valueSchema.decode(valueBytes, schemaVersion);
+ if (valueSchema.supportSchemaVersioning() && schemaIdOrVersion != null) {
+ byte[] valueSchemaIdOrVersion = schemaIdOrVersion;
+ if (SchemaType.EXTERNAL.equals(valueSchema.getSchemaInfo().getType())) {
+ valueSchemaIdOrVersion = KeyValue.getSchemaId(schemaIdOrVersion, false);
+ }
+ v = valueSchema.decode(topic, valueBytes, valueSchemaIdOrVersion);
Review Comment:
fixed
##########
pulsar-common/src/main/proto/PulsarApi.proto:
##########
@@ -173,6 +174,7 @@ message MessageMetadata {
// the `compacted_out` field in `SingleMessageMetadata` must be checked to identify and filter out compacted
// messages (e.g., `k1 => v1` and `k1 => null` in the example above).
repeated int32 compacted_batch_indexes = 31;
+ optional bytes schema_id = 32;
Review Comment:
updated PIP-420
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306785767
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java:
##########
@@ -375,7 +375,7 @@ public void testSeparatedKeyValueEncodingTypeSchemaEncodeAndDecode() {
} catch (SchemaSerializationException e) {
Assert.assertTrue(e.getMessage().contains("This method cannot be used under this SEPARATED encoding type"));
}
- KeyValue<Foo, Bar> keyValue = ((KeyValueSchemaImpl) keyValueSchema).decode(fooSchema.encode(foo),
+ KeyValue<Foo, Bar> keyValue = ((KeyValueSchemaImpl) keyValueSchema).decode(null, fooSchema.encode(foo),
Review Comment:
fixed
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java:
##########
@@ -717,4 +717,5 @@ ClientBuilder authentication(String authPluginClassName, Map<String, String> aut
* - The `loadManagerClassName` config in broker is a class that implements the `ExtensibleLoadManager` interface
*/
ClientBuilder lookupProperties(Map<String, String> properties);
+
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306819935
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -424,6 +424,23 @@ public byte[] getSchemaVersion() {
}
}
+ @Override
+ public byte[] getSchemaId() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
+ return null;
+ }
+
+ private byte[] getSchemaFlag() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
Review Comment:
Sure, I can add a new method for decoding with schemaId. We added two decode methods in the `Schema`, they should be only used for decoding with the schema ID
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297399483
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
Review Comment:
The schema compatibility check between two `EXTERNAL` schemas should be done in the external schema registry, so it just do nothing if `from` and `to` are both `EXTERNAL`
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297442619
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java:
##########
@@ -184,6 +197,10 @@ default void configureSchemaInfo(String topic, String componentName,
*/
Schema<T> clone();
+ default CompletableFuture<Void> closeAsync() {
+ return CompletableFuture.completedFuture(null);
+ }
Review Comment:
Seems this method only has an default implementation.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#issuecomment-3219143474
@Denovo1998 Any chance to take a look again?
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297440191
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/EncodeData.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.api;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@AllArgsConstructor
+@Data
+public class EncodeData {
+
+ private byte[] data;
+
+ private byte[] schemaId;
+
+ public EncodeData(byte[] data) {
+ this.data = data;
+ this.schemaId = new byte[0];
+ }
Review Comment:
It's better to remove this constructor. I see `new EncodeData(data, null)` is called in the default `Schema#encode(String topic, T message)` implementation.
Mixing the use cases of `null` and `new byte[0]` makes code hard to read. Actually the immutable `record` should be best here.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299681668
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -424,6 +424,23 @@ public byte[] getSchemaVersion() {
}
}
+ @Override
+ public byte[] getSchemaId() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
+ return null;
+ }
+
+ private byte[] getSchemaFlag() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
Review Comment:
Yes, the schema ID should only be used to decode the message payload.
- Changed get schemaInfo to use the schema version.
- Changed get schema reader to use the schema version.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2295427146
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java:
##########
@@ -71,6 +71,7 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
} catch (InvalidSchemaDataException invalidSchemaDataException) {
throw invalidSchemaDataException;
} catch (Exception e) {
+ e.printStackTrace();
Review Comment:
```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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2269493366
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if (strategy == SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE) {
+ return;
+ }
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
+ }
+
+ @Override
+ public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if (strategy == SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE) {
+ return;
+ }
+ while (from.iterator().hasNext()) {
+ SchemaData fromSchema = from.iterator().next();
+ checkCompatible(fromSchema, to, strategy);
+ }
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 closed pull request #24488: [improve][client] PIP-420: Supports users implement external schemas
URL: https://github.com/apache/pulsar/pull/24488
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306780392
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,28 +101,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (keySchemaId == null && valueSchemaId == null) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ byteBuffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
return byteBuffer.array();
}
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
+ ByteBuffer buffer = ByteBuffer.wrap(schemaId);
+ int keySchemaLength = buffer.getInt();
+ if (keySchemaLength > 0) {
+ byte[] keySchemaId = new byte[keySchemaLength];
+ buffer.get(keySchemaId);
+ if (isKey) {
+ return keySchemaId;
+ }
+ }
+ int valueSchemaLength = buffer.getInt();
+ if (valueSchemaLength > 0) {
+ byte[] valueSchemaId = new byte[valueSchemaLength];
+ buffer.get(valueSchemaId);
+ return valueSchemaId;
+ }
+ throw new IllegalArgumentException("Invalid schemaId length: " + schemaId.length);
Review Comment:
Improved, and I added a test to cover different cases. https://github.com/apache/pulsar/pull/24488/files#diff-2c0d48104dc0a3b48606a351a0af057d337274afd61e96ebc8bd0aabee3c8a18R152
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,28 +101,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (keySchemaId == null && valueSchemaId == null) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ byteBuffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
return byteBuffer.array();
}
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
Review Comment:
Improved, and I added a test to cover different cases. https://github.com/apache/pulsar/pull/24488/files#diff-2c0d48104dc0a3b48606a351a0af057d337274afd61e96ebc8bd0aabee3c8a18R152
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2300765662
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaImpl.java:
##########
@@ -158,26 +180,40 @@ public KeyValue<K, V> decode(byte[] bytes, byte[] schemaVersion) {
if (this.keyValueEncodingType == KeyValueEncodingType.SEPARATED) {
throw new SchemaSerializationException("This method cannot be used under this SEPARATED encoding type");
}
- return KeyValue.decode(bytes, (keyBytes, valueBytes) -> decode(keyBytes, valueBytes, schemaVersion));
+ return KeyValue.decode(bytes, (keyBytes, valueBytes) ->
+ decode(null, keyBytes, valueBytes, schemaVersion));
}
@Override
public KeyValue<K, V> decode(ByteBuf byteBuf) {
return decode(ByteBufUtil.getBytes(byteBuf));
}
+ @Override
+ public KeyValue<K, V> decode(String topic, byte[] data, byte[] schemaId) {
+ if (this.keyValueEncodingType == KeyValueEncodingType.SEPARATED) {
+ throw new SchemaSerializationException("This method cannot be used under this SEPARATED encoding type");
+ }
+ return KeyValue.decode(data, (keyBytes, valueBytes) ->
+ decode(topic, keyBytes, valueBytes, schemaId));
+ }
+
@Override
public KeyValue<K, V> decode(ByteBuf byteBuf, byte[] schemaVersion) {
return decode(ByteBufUtil.getBytes(byteBuf), schemaVersion);
}
- public KeyValue<K, V> decode(byte[] keyBytes, byte[] valueBytes, byte[] schemaVersion) {
+ public KeyValue<K, V> decode(String topic, byte[] keyBytes, byte[] valueBytes, byte[] schemaIdOrVersion) {
K k;
if (keyBytes == null) {
k = null;
} else {
- if (keySchema.supportSchemaVersioning() && schemaVersion != null) {
- k = keySchema.decode(keyBytes, schemaVersion);
+ if (keySchema.supportSchemaVersioning() && schemaIdOrVersion != null) {
+ byte[] keySchemaIdOrVersion = schemaIdOrVersion;
+ if (SchemaType.EXTERNAL.equals(keySchema.getSchemaInfo().getType())) {
+ keySchemaIdOrVersion = KeyValue.getSchemaId(schemaIdOrVersion, true);
+ }
+ k = keySchema.decode(topic, keyBytes, keySchemaIdOrVersion);
Review Comment:
Summarize
```java
K k = null;
if (keyBytes != null) {
if (keySchema.supportSchemaVersioning() && schemaIdOrVersion != null) {
byte[] keySchemaIdOrVersion = schemaIdOrVersion;
if (SchemaType.EXTERNAL.equals(keySchema.getSchemaInfo().getType())) {
keySchemaIdOrVersion = KeyValue.getSchemaId(schemaIdOrVersion, true);
}
k = keySchema.decode(topic, keyBytes, keySchemaIdOrVersion);
} else {
k = keySchema.decode(keyBytes);
}
}
```
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ClientBuilder.java:
##########
@@ -717,4 +717,5 @@ ClientBuilder authentication(String authPluginClassName, Map<String, String> aut
* - The `loadManagerClassName` config in broker is a class that implements the `ExtensibleLoadManager` interface
*/
ClientBuilder lookupProperties(Map<String, String> properties);
+
Review Comment:
Unnecessary modifications.
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java:
##########
@@ -375,7 +375,7 @@ public void testSeparatedKeyValueEncodingTypeSchemaEncodeAndDecode() {
} catch (SchemaSerializationException e) {
Assert.assertTrue(e.getMessage().contains("This method cannot be used under this SEPARATED encoding type"));
}
- KeyValue<Foo, Bar> keyValue = ((KeyValueSchemaImpl) keyValueSchema).decode(fooSchema.encode(foo),
+ KeyValue<Foo, Bar> keyValue = ((KeyValueSchemaImpl) keyValueSchema).decode(null, fooSchema.encode(foo),
Review Comment:
Here are two spaces
```java
KeyValue<Foo, Bar> keyValue
```
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionSchemaInfoProvider.java:
##########
@@ -100,4 +100,5 @@ private CompletableFuture<SchemaInfo> loadSchema(byte[] schemaVersion) {
public PulsarClientImpl getPulsarClient() {
return pulsarClient;
}
+
Review Comment:
Unnecessary modifications.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java:
##########
@@ -505,4 +505,5 @@ public ClientBuilder lookupProperties(Map<String, String> properties) {
conf.setLookupProperties(properties);
return this;
}
+
Review Comment:
Unnecessary modifications.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java:
##########
@@ -508,4 +508,5 @@ public void setLookupProperties(Map<String, String> lookupProperties) {
public Map<String, String> getLookupProperties() {
return (lookupProperties == null) ? Collections.emptyMap() : Collections.unmodifiableMap(lookupProperties);
}
+
Review Comment:
Unnecessary modifications.
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java:
##########
@@ -375,7 +375,7 @@ public void testSeparatedKeyValueEncodingTypeSchemaEncodeAndDecode() {
} catch (SchemaSerializationException e) {
Assert.assertTrue(e.getMessage().contains("This method cannot be used under this SEPARATED encoding type"));
}
- KeyValue<Foo, Bar> keyValue = ((KeyValueSchemaImpl) keyValueSchema).decode(fooSchema.encode(foo),
+ KeyValue<Foo, Bar> keyValue = ((KeyValueSchemaImpl) keyValueSchema).decode(null, fooSchema.encode(foo),
Review Comment:
It seems there are also two spaces in the `testInlineKeyValueEncodingTypeSchemaEncodeAndDecode` above.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaImpl.java:
##########
@@ -187,8 +223,12 @@ public KeyValue<K, V> decode(byte[] keyBytes, byte[] valueBytes, byte[] schemaVe
if (valueBytes == null) {
v = null;
} else {
- if (valueSchema.supportSchemaVersioning() && schemaVersion != null) {
- v = valueSchema.decode(valueBytes, schemaVersion);
+ if (valueSchema.supportSchemaVersioning() && schemaIdOrVersion != null) {
+ byte[] valueSchemaIdOrVersion = schemaIdOrVersion;
+ if (SchemaType.EXTERNAL.equals(valueSchema.getSchemaInfo().getType())) {
+ valueSchemaIdOrVersion = KeyValue.getSchemaId(schemaIdOrVersion, false);
+ }
+ v = valueSchema.decode(topic, valueBytes, valueSchemaIdOrVersion);
Review Comment:
Summarize
```java
V v = null;
if (valueBytes != null) {
if (valueSchema.supportSchemaVersioning() && schemaIdOrVersion != null) {
byte[] valueSchemaIdOrVersion = schemaIdOrVersion;
if (SchemaType.EXTERNAL.equals(valueSchema.getSchemaInfo().getType())) {
valueSchemaIdOrVersion = KeyValue.getSchemaId(schemaIdOrVersion, false);
}
v = valueSchema.decode(topic, valueBytes, valueSchemaIdOrVersion);
} else {
v = valueSchema.decode(valueBytes);
}
}
```
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2309244211
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,26 +102,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
- return byteBuffer.array();
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (!isValidSchemaId(keySchemaId) && !isValidSchemaId(valueSchemaId)) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer buffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ buffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
+ return buffer.array();
+ }
+
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
+ ByteBuffer buffer = ByteBuffer.wrap(schemaId);
+ int keySchemaLength = buffer.getInt();
+ byte[] keySchemaId = new byte[0];
+ if (keySchemaLength > 0) {
+ keySchemaId = new byte[keySchemaLength];
+ buffer.get(keySchemaId);
+ }
+ if (isKey) {
Review Comment:
If `isKey` is true, we can just skip the key bytes like:
```java
if (isKey) {
// ...
} else {
int keySchemaLength = buffer.getInt();
buffer.position(buffer.position() + keySchemaLength);
// TODO:
}
```
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299590642
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheck.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+
+public class ExternalSchemaCompatibilityCheck implements SchemaCompatibilityCheck {
+
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.EXTERNAL;
+ }
+
+ @Override
+ public void checkCompatible(SchemaData from, SchemaData to, SchemaCompatibilityStrategy strategy)
+ throws IncompatibleSchemaException {
+ if ((SchemaType.EXTERNAL.equals(from.getType()) || SchemaType.EXTERNAL.equals(to.getType()))
+ && !from.getType().equals(to.getType())) {
+ throw new IncompatibleSchemaException("External schema is not compatible with the other schema types.");
+ }
Review Comment:
In any strategy mode, the external schema is not compatible with other Pulsar schemas; the external schema manager will manage the external schema compatibility check.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306780392
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,28 +101,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (keySchemaId == null && valueSchemaId == null) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ byteBuffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
return byteBuffer.array();
}
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
+ ByteBuffer buffer = ByteBuffer.wrap(schemaId);
+ int keySchemaLength = buffer.getInt();
+ if (keySchemaLength > 0) {
+ byte[] keySchemaId = new byte[keySchemaLength];
+ buffer.get(keySchemaId);
+ if (isKey) {
+ return keySchemaId;
+ }
+ }
+ int valueSchemaLength = buffer.getInt();
+ if (valueSchemaLength > 0) {
+ byte[] valueSchemaId = new byte[valueSchemaLength];
+ buffer.get(valueSchemaId);
+ return valueSchemaId;
+ }
+ throw new IllegalArgumentException("Invalid schemaId length: " + schemaId.length);
Review Comment:
Improved, and I added a test to cover different cases.
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,28 +101,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (keySchemaId == null && valueSchemaId == null) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ byteBuffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
return byteBuffer.array();
}
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
Review Comment:
Improved, and I added a test to cover different cases.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306365993
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -424,6 +424,23 @@ public byte[] getSchemaVersion() {
}
}
+ @Override
+ public byte[] getSchemaId() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
+ return null;
+ }
+
+ private byte[] getSchemaFlag() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
Review Comment:
I still think we should not mix schema id and version even in decoding. For example, you changed
```java
return decode(schema.supportSchemaVersioning() ? getSchemaIVersion() : null);
```
to
```java
return decode(schema.supportSchemaVersioning() ? getSchemaIdOrVersion() : null);
```
It changes the semantics of the `supportSchemaVersioning` method that if it's true, schema id could be used to decode the payload, which is unexpected.
Why not just keep the original logic not changed and add a separated method to decode value by schema id?
```java
@Override
public T getValue() {
if (msgMetadata.hasSchemaId()) {
return decodeBySchemaId(msgMetadata.getSchemaId());
}
// ...
```
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297362285
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Message.java:
##########
@@ -208,6 +208,15 @@ public interface Message<T> {
*/
byte[] getSchemaVersion();
+ /**
+ * Get schema ID of the message.
+ * PIP-420 provides a way to produce messages with external schema,
+ * and the schema ID will be set to the message metadata.
+ *
+ * @return schema ID of the message if the message is produced with external schema.
Review Comment:
Please note `null` value can be returned if it does not have a schema id.
I also noticed an empty array also represents "no schema id". Please make the behavior clear in documents.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#issuecomment-3197486357
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24488?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `74.81481%` with `34 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 74.27%. Comparing base ([`19b7c27`](https://app.codecov.io/gh/apache/pulsar/commit/19b7c2765c56794a6926603c4d4a11c39c21faa6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8f031d2`](https://app.codecov.io/gh/apache/pulsar/commit/8f031d255f86399f20562045775ac0bbe75e2c81?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 2 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24488?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../pulsar/client/impl/schema/KeyValueSchemaImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Fschema%2FKeyValueSchemaImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL3NjaGVtYS9LZXlWYWx1ZVNjaGVtYUltcGwuamF2YQ==) | 60.86% | [5 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rvice/schema/ExternalSchemaCompatibilityCheck.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fschema%2FExternalSchemaCompatibilityCheck.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3NjaGVtYS9FeHRlcm5hbFNjaGVtYUNvbXBhdGliaWxpdHlDaGVjay5qYXZh) | 30.00% | [5 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../pulsar/client/impl/BatchMessageContainerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FBatchMessageContainerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0JhdGNoTWVzc2FnZUNvbnRhaW5lckltcGwuamF2YQ==) | 20.00% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ava/org/apache/pulsar/client/impl/MessageImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FMessageImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL01lc3NhZ2VJbXBsLmphdmE=) | 88.00% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...he/pulsar/client/impl/TypedMessageBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FTypedMessageBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1R5cGVkTWVzc2FnZUJ1aWxkZXJJbXBsLmphdmE=) | 80.00% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...va/org/apache/pulsar/client/impl/ConsumerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FConsumerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NvbnN1bWVySW1wbC5qYXZh) | 60.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...va/org/apache/pulsar/client/impl/ProducerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FProducerImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1Byb2R1Y2VySW1wbC5qYXZh) | 80.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rg/apache/pulsar/client/impl/TopicMessageImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FTopicMessageImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1RvcGljTWVzc2FnZUltcGwuamF2YQ==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ce/schema/validator/StructSchemaDataValidator.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fschema%2Fvalidator%2FStructSchemaDataValidator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3NjaGVtYS92YWxpZGF0b3IvU3RydWN0U2NoZW1hRGF0YVZhbGlkYXRvci5qYXZh) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../pulsar/client/impl/schema/KeyValueSchemaInfo.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Fschema%2FKeyValueSchemaInfo.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL3NjaGVtYS9LZXlWYWx1ZVNjaGVtYUluZm8uamF2YQ==) | 50.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24488 +/- ##
============================================
- Coverage 74.48% 74.27% -0.22%
+ Complexity 33277 32822 -455
============================================
Files 1882 1884 +2
Lines 146854 146947 +93
Branches 16866 16887 +21
============================================
- Hits 109389 109142 -247
- Misses 28861 29121 +260
- Partials 8604 8684 +80
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24488/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24488/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.72% <17.77%> (+0.02%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24488/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.32% <52.59%> (-0.06%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24488/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.75% <74.81%> (-0.23%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24488?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...org/apache/pulsar/broker/ServiceConfiguration.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-broker-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2FServiceConfiguration.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9icm9rZXIvU2VydmljZUNvbmZpZ3VyYXRpb24uamF2YQ==) | `98.17% <ø> (ø)` | |
| [.../service/schema/validator/SchemaDataValidator.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fschema%2Fvalidator%2FSchemaDataValidator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3NjaGVtYS92YWxpZGF0b3IvU2NoZW1hRGF0YVZhbGlkYXRvci5qYXZh) | `80.00% <ø> (ø)` | |
| [.../java/org/apache/pulsar/client/api/EncodeData.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fapi%2FEncodeData.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jbGllbnQvYXBpL0VuY29kZURhdGEuamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [...ain/java/org/apache/pulsar/client/api/Message.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fapi%2FMessage.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jbGllbnQvYXBpL01lc3NhZ2UuamF2YQ==) | `0.00% <ø> (ø)` | |
| [...main/java/org/apache/pulsar/client/api/Schema.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fapi%2FSchema.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jbGllbnQvYXBpL1NjaGVtYS5qYXZh) | `83.33% <100.00%> (+1.19%)` | :arrow_up: |
| [...java/org/apache/pulsar/common/schema/KeyValue.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Fschema%2FKeyValue.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jb21tb24vc2NoZW1hL0tleVZhbHVlLmphdmE=) | `91.07% <100.00%> (+2.97%)` | :arrow_up: |
| [...va/org/apache/pulsar/common/schema/SchemaType.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client-api%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fcommon%2Fschema%2FSchemaType.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3B1bHNhci9jb21tb24vc2NoZW1hL1NjaGVtYVR5cGUuamF2YQ==) | `83.58% <100.00%> (+0.50%)` | :arrow_up: |
| [...g/apache/pulsar/client/impl/ClientBuilderImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2FClientBuilderImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL0NsaWVudEJ1aWxkZXJJbXBsLmphdmE=) | `85.79% <ø> (ø)` | |
| [...lsar/client/impl/conf/ClientConfigurationData.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Fconf%2FClientConfigurationData.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL2NvbmYvQ2xpZW50Q29uZmlndXJhdGlvbkRhdGEuamF2YQ==) | `93.33% <ø> (ø)` | |
| [...schema/generic/MultiVersionSchemaInfoProvider.java](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree&filepath=pulsar-client%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fclient%2Fimpl%2Fschema%2Fgeneric%2FMultiVersionSchemaInfoProvider.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL3NjaGVtYS9nZW5lcmljL011bHRpVmVyc2lvblNjaGVtYUluZm9Qcm92aWRlci5qYXZh) | `70.00% <ø> (ø)` | |
| ... and [11 more](https://app.codecov.io/gh/apache/pulsar/pull/24488?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [87 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24488/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower merged PR #24488:
URL: https://github.com/apache/pulsar/pull/24488
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2296924769
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java:
##########
@@ -71,6 +71,7 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
} catch (InvalidSchemaDataException invalidSchemaDataException) {
throw invalidSchemaDataException;
} catch (Exception e) {
+ e.printStackTrace();
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297456892
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -501,16 +518,16 @@ private T decode(byte[] schemaVersion) {
}
}
- private T decodeBySchema(byte[] schemaVersion) {
- T value = poolMessage ? schema.decode(payload.nioBuffer(), schemaVersion) : null;
+ private T decodeBySchema(byte[] schemaFlag) {
Review Comment:
Should it be renamed with `schemaId`? The `schemaFlag` concept in code is confusing. I see the result of `getSchemaFlag()` is still passed to `getSchemaInfo` method, which accepts a schema version
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299681668
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java:
##########
@@ -424,6 +424,23 @@ public byte[] getSchemaVersion() {
}
}
+ @Override
+ public byte[] getSchemaId() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
+ return null;
+ }
+
+ private byte[] getSchemaFlag() {
+ if (msgMetadata.hasSchemaId()) {
+ byte[] schemaId = msgMetadata.getSchemaId();
+ return (schemaId.length == 0) ? null : schemaId;
+ }
Review Comment:
Yes, you are right, the schema ID should only be used to decode the message payload.
- Changed get schemaInfo to use the schema version.
- Changed get schema reader to use the schema version.
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2301147448
##########
pulsar-client-api/src/main/java/org/apache/pulsar/common/schema/KeyValue.java:
##########
@@ -100,28 +101,82 @@ public interface KeyValueDecoder<K, V> {
*/
public static <K, V> byte[] encode(K key, Schema<K> keyWriter,
V value, Schema<V> valueWriter) {
- byte [] keyBytes;
+ return encode(null, key, keyWriter, value, valueWriter).data();
+ }
+
+ public static <K, V> EncodeData encode(String topic, K key, Schema<K> keyWriter,
+ V value, Schema<V> valueWriter) {
+ EncodeData keyEncodeData;
if (key == null) {
- keyBytes = new byte[0];
+ keyEncodeData = new EncodeData(new byte[0]);
} else {
- keyBytes = keyWriter.encode(key);
+ keyEncodeData = keyWriter.encode(topic, key);
}
- byte [] valueBytes;
+ EncodeData valueEncodeData;
if (value == null) {
- valueBytes = new byte[0];
+ valueEncodeData = new EncodeData(new byte[0]);
} else {
- valueBytes = valueWriter.encode(value);
+ valueEncodeData = valueWriter.encode(topic, value);
}
- ByteBuffer byteBuffer = ByteBuffer.allocate(4 + keyBytes.length + 4 + valueBytes.length);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keyEncodeData.data().length + 4 + valueEncodeData.data().length);
byteBuffer
- .putInt(key == null ? -1 : keyBytes.length)
- .put(keyBytes)
- .putInt(value == null ? -1 : valueBytes.length)
- .put(valueBytes);
+ .putInt(key == null ? -1 : keyEncodeData.data().length)
+ .put(keyEncodeData.data())
+ .putInt(value == null ? -1 : valueEncodeData.data().length)
+ .put(valueEncodeData.data());
+ return new EncodeData(byteBuffer.array(),
+ generateKVSchemaId(keyEncodeData.schemaId(), valueEncodeData.schemaId()));
+ }
+
+ /**
+ * Generate a combined schema id for key/value schema.
+ * The format is:
+ * schemaId = schemaKeyLength + keySchemaIdBytes + schemaValueLength + valueSchemaIdBytes
+ * where schemaKeyLength and schemaValueLength are 4 bytes integer.
+ * If keySchemaIdBytes or valueSchemaIdBytes is null, the length will be 0.
+ * So the total length of schemaId is:
+ * 4 + keySchemaIdBytes.length + 4 + valueSchemaIdBytes.length
+ *
+ * @param keySchemaId the schema id of key schema
+ * @param valueSchemaId the schema id of value schema
+ */
+ public static byte[] generateKVSchemaId(byte[] keySchemaId, byte[] valueSchemaId) {
+ if (keySchemaId == null && valueSchemaId == null) {
+ return null;
+ }
+ keySchemaId = keySchemaId == null ? new byte[0] : keySchemaId;
+ valueSchemaId = valueSchemaId == null ? new byte[0] : valueSchemaId;
+ ByteBuffer byteBuffer = ByteBuffer.allocate(
+ 4 + keySchemaId.length + 4 + valueSchemaId.length);
+ byteBuffer
+ .putInt(keySchemaId.length)
+ .put(keySchemaId)
+ .putInt(valueSchemaId.length)
+ .put(valueSchemaId);
return byteBuffer.array();
}
+ public static byte[] getSchemaId(byte[] schemaId, boolean isKey) {
+ ByteBuffer buffer = ByteBuffer.wrap(schemaId);
+ int keySchemaLength = buffer.getInt();
+ if (keySchemaLength > 0) {
+ byte[] keySchemaId = new byte[keySchemaLength];
+ buffer.get(keySchemaId);
+ if (isKey) {
+ return keySchemaId;
+ }
+ }
+ int valueSchemaLength = buffer.getInt();
+ if (valueSchemaLength > 0) {
+ byte[] valueSchemaId = new byte[valueSchemaLength];
+ buffer.get(valueSchemaId);
+ return valueSchemaId;
+ }
+ throw new IllegalArgumentException("Invalid schemaId length: " + schemaId.length);
Review Comment:
Is valueSchema not allowed to be null?
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2307002689
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheckTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import static org.testng.Assert.fail;
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ExternalSchemaCompatibilityCheckTest {
+
+ private final ExternalSchemaCompatibilityCheck compatibilityCheck = new ExternalSchemaCompatibilityCheck();
+
+ private final SchemaData externalSchemaData = SchemaData.builder()
+ .type(SchemaType.EXTERNAL)
+ .data(new byte[0])
+ .build();
+
+ @DataProvider(name = "otherSchemasProvider")
+ public Object[] otherSchemasProvider() {
+ return new Object[] {
+ SchemaData.builder()
+ .type(SchemaType.JSON)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.AVRO)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF_NATIVE)
+ .build()
+ };
+ }
+
+ @Test(dataProvider = "otherSchemasProvider")
+ public void testExternalSchemaCompatibilityCheck(SchemaData schemaData) {
Review Comment:
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2300675073
##########
pulsar-common/src/main/proto/PulsarApi.proto:
##########
@@ -173,6 +174,7 @@ message MessageMetadata {
// the `compacted_out` field in `SingleMessageMetadata` must be checked to identify and filter out compacted
// messages (e.g., `k1 => v1` and `k1 => null` in the example above).
repeated int32 compacted_batch_indexes = 31;
+ optional bytes schema_id = 32;
Review Comment:
It looks like it says `31` in the PIP documentation, should we update the PIP documentation in 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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2299604142
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheckTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import static org.testng.Assert.fail;
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ExternalSchemaCompatibilityCheckTest {
+
+ private final ExternalSchemaCompatibilityCheck compatibilityCheck = new ExternalSchemaCompatibilityCheck();
+
+ private final SchemaData externalSchemaData = SchemaData.builder()
+ .type(SchemaType.EXTERNAL)
+ .data(new byte[0])
+ .build();
+
+ @DataProvider(name = "otherSchemasProvider")
+ public Object[] otherSchemasProvider() {
+ return new Object[] {
+ SchemaData.builder()
+ .type(SchemaType.JSON)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.AVRO)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF_NATIVE)
+ .build()
+ };
+ }
+
+ @Test(dataProvider = "otherSchemasProvider")
+ public void testExternalSchemaCompatibilityCheck(SchemaData schemaData) {
Review Comment:
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "gaoran10 (via GitHub)" <gi...@apache.org>.
gaoran10 commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2306786338
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java:
##########
@@ -508,4 +508,5 @@ public void setLookupProperties(Map<String, String> lookupProperties) {
public Map<String, String> getLookupProperties() {
return (lookupProperties == null) ? Collections.emptyMap() : Collections.unmodifiableMap(lookupProperties);
}
+
Review Comment:
fixed
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionSchemaInfoProvider.java:
##########
@@ -100,4 +100,5 @@ private CompletableFuture<SchemaInfo> loadSchema(byte[] schemaVersion) {
public PulsarClientImpl getPulsarClient() {
return pulsarClient;
}
+
Review Comment:
fixed
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaImpl.java:
##########
@@ -158,26 +180,40 @@ public KeyValue<K, V> decode(byte[] bytes, byte[] schemaVersion) {
if (this.keyValueEncodingType == KeyValueEncodingType.SEPARATED) {
throw new SchemaSerializationException("This method cannot be used under this SEPARATED encoding type");
}
- return KeyValue.decode(bytes, (keyBytes, valueBytes) -> decode(keyBytes, valueBytes, schemaVersion));
+ return KeyValue.decode(bytes, (keyBytes, valueBytes) ->
+ decode(null, keyBytes, valueBytes, schemaVersion));
}
@Override
public KeyValue<K, V> decode(ByteBuf byteBuf) {
return decode(ByteBufUtil.getBytes(byteBuf));
}
+ @Override
+ public KeyValue<K, V> decode(String topic, byte[] data, byte[] schemaId) {
+ if (this.keyValueEncodingType == KeyValueEncodingType.SEPARATED) {
+ throw new SchemaSerializationException("This method cannot be used under this SEPARATED encoding type");
+ }
+ return KeyValue.decode(data, (keyBytes, valueBytes) ->
+ decode(topic, keyBytes, valueBytes, schemaId));
+ }
+
@Override
public KeyValue<K, V> decode(ByteBuf byteBuf, byte[] schemaVersion) {
return decode(ByteBufUtil.getBytes(byteBuf), schemaVersion);
}
- public KeyValue<K, V> decode(byte[] keyBytes, byte[] valueBytes, byte[] schemaVersion) {
+ public KeyValue<K, V> decode(String topic, byte[] keyBytes, byte[] valueBytes, byte[] schemaIdOrVersion) {
K k;
if (keyBytes == null) {
k = null;
} else {
- if (keySchema.supportSchemaVersioning() && schemaVersion != null) {
- k = keySchema.decode(keyBytes, schemaVersion);
+ if (keySchema.supportSchemaVersioning() && schemaIdOrVersion != null) {
+ byte[] keySchemaIdOrVersion = schemaIdOrVersion;
+ if (SchemaType.EXTERNAL.equals(keySchema.getSchemaInfo().getType())) {
+ keySchemaIdOrVersion = KeyValue.getSchemaId(schemaIdOrVersion, true);
+ }
+ k = keySchema.decode(topic, keyBytes, keySchemaIdOrVersion);
Review Comment:
fixed
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientBuilderImpl.java:
##########
@@ -505,4 +505,5 @@ public ClientBuilder lookupProperties(Map<String, String> properties) {
conf.setLookupProperties(properties);
return this;
}
+
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297402084
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ExternalSchemaCompatibilityCheckTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.schema;
+
+import static org.testng.Assert.fail;
+import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
+import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
+import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class ExternalSchemaCompatibilityCheckTest {
+
+ private final ExternalSchemaCompatibilityCheck compatibilityCheck = new ExternalSchemaCompatibilityCheck();
+
+ private final SchemaData externalSchemaData = SchemaData.builder()
+ .type(SchemaType.EXTERNAL)
+ .data(new byte[0])
+ .build();
+
+ @DataProvider(name = "otherSchemasProvider")
+ public Object[] otherSchemasProvider() {
+ return new Object[] {
+ SchemaData.builder()
+ .type(SchemaType.JSON)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.AVRO)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF)
+ .build(),
+ SchemaData.builder()
+ .type(SchemaType.PROTOBUF_NATIVE)
+ .build()
+ };
+ }
+
+ @Test(dataProvider = "otherSchemasProvider")
+ public void testExternalSchemaCompatibilityCheck(SchemaData schemaData) {
Review Comment:
Could you test `checkCompatible(externalSchemaData, schemaData` as well?
--
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][client] PIP-420: Supports users implement external schemas [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #24488:
URL: https://github.com/apache/pulsar/pull/24488#discussion_r2297482086
##########
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java:
##########
@@ -184,6 +197,10 @@ default void configureSchemaInfo(String topic, String componentName,
*/
Schema<T> clone();
+ default CompletableFuture<Void> closeAsync() {
+ return CompletableFuture.completedFuture(null);
+ }
Review Comment:
Because built-in `Schema` implementations don't need to implement `closeAsync`. This method is added to support closing the connection to an external schema registry service for an external `Schema` implementation.
--
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