diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java index 70554926c9dc0..9fcd231599631 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/ClientGetSchemaTest.java @@ -31,6 +31,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -43,6 +44,8 @@ public class ClientGetSchemaTest extends ProducerConsumerBase { private static final String topicString = "my-property/my-ns/topic-string"; private static final String topicJson = "my-property/my-ns/topic-json"; private static final String topicAvro = "my-property/my-ns/topic-avro"; + private static final String topicJsonNotNull = "my-property/my-ns/topic-json-not-null"; + private static final String topicAvroNotNull = "my-property/my-ns/topic-avro-not-null"; List> producers = new ArrayList<>(); @@ -62,6 +65,11 @@ protected void setup() throws Exception { producers.add(pulsarClient.newProducer(Schema.STRING).topic(topicString).create()); producers.add(pulsarClient.newProducer(Schema.AVRO(MyClass.class)).topic(topicAvro).create()); producers.add(pulsarClient.newProducer(Schema.JSON(MyClass.class)).topic(topicJson).create()); + producers.add(pulsarClient.newProducer(Schema.AVRO(SchemaDefinition.builder().withPojo(MyClass.class).build())).topic(topicAvro).create()); + producers.add(pulsarClient.newProducer(Schema.JSON(SchemaDefinition.builder().withPojo(MyClass.class).build())).topic(topicJson).create()); + producers.add(pulsarClient.newProducer(Schema.AVRO(SchemaDefinition.builder().withPojo(MyClass.class).withAlwaysAllowNull(false).build())).topic(topicAvroNotNull).create()); + producers.add(pulsarClient.newProducer(Schema.JSON(SchemaDefinition.builder().withPojo(MyClass.class).withAlwaysAllowNull(false).build())).topic(topicJsonNotNull).create()); + } @AfterClass diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java index 17b39b79677cc..a78592740bfdc 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java @@ -33,6 +33,7 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.common.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaInfo; @@ -51,11 +52,11 @@ public SchemaCompatibilityCheck getSchemaCheck() { public void testJsonSchemaBackwardsCompatibility() throws JsonProcessingException { SchemaData from = SchemaData.builder().data(OldJSONSchema.of(Foo.class).getSchemaInfo().getSchema()).build(); - SchemaData to = SchemaData.builder().data(JSONSchema.of(Foo.class).getSchemaInfo().getSchema()).build(); + SchemaData to = SchemaData.builder().data(JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()).getSchemaInfo().getSchema()).build(); JsonSchemaCompatibilityCheck jsonSchemaCompatibilityCheck = new JsonSchemaCompatibilityCheck(); Assert.assertTrue(jsonSchemaCompatibilityCheck.isCompatible(from, to, SchemaCompatibilityStrategy.FULL)); - from = SchemaData.builder().data(JSONSchema.of(Foo.class).getSchemaInfo().getSchema()).build(); + from = SchemaData.builder().data(JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()).getSchemaInfo().getSchema()).build(); to = SchemaData.builder().data(OldJSONSchema.of(Foo.class).getSchemaInfo().getSchema()).build(); Assert.assertTrue(jsonSchemaCompatibilityCheck.isCompatible(from, to, SchemaCompatibilityStrategy.FULL)); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java index c9e919e06794e..050db394fb0e5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java @@ -30,6 +30,7 @@ import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.service.schema.SchemaRegistry; import org.apache.pulsar.client.api.schema.GenericRecord; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.client.impl.schema.ProtobufSchema; @@ -63,7 +64,7 @@ public void testJsonProducerAndConsumer() throws Exception { log.info("-- Starting {} test --", methodName); JSONSchema jsonSchema = - JSONSchema.of(JsonEncodedPojo.class); + JSONSchema.of(SchemaDefinition.builder().withPojo(JsonEncodedPojo.class).build()); Consumer consumer = pulsarClient .newConsumer(jsonSchema) @@ -108,7 +109,7 @@ public void testJsonProducerAndConsumerWithPrestoredSchema() throws Exception { log.info("-- Starting {} test --", methodName); JSONSchema jsonSchema = - JSONSchema.of(JsonEncodedPojo.class); + JSONSchema.of(SchemaDefinition.builder().withPojo(JsonEncodedPojo.class).build()); pulsar.getSchemaRegistryService() .putSchemaIfAbsent("my-property/my-ns/my-topic1", @@ -166,7 +167,7 @@ public void testJsonConsumerWithWrongCorruptedSchema() throws Exception { ).get(); Consumer consumer = pulsarClient - .newConsumer(JSONSchema.of(JsonEncodedPojo.class)) + .newConsumer(JSONSchema.of(SchemaDefinition.builder().withPojo(JsonEncodedPojo.class).build())) .topic("persistent://my-property/use/my-ns/my-topic1") .subscriptionName("my-subscriber-name") .subscribe(); @@ -194,7 +195,7 @@ public void testJsonProducerWithWrongCorruptedSchema() throws Exception { ).get(); Producer producer = pulsarClient - .newProducer(JSONSchema.of(JsonEncodedPojo.class)) + .newProducer(JSONSchema.of(SchemaDefinition.builder().withPojo(JsonEncodedPojo.class).build())) .topic("persistent://my-property/use/my-ns/my-topic1") .create(); @@ -273,7 +274,9 @@ public void testProtobufConsumerWithWrongPrestoredSchema() throws Exception { ).get(); Consumer consumer = pulsarClient - .newConsumer(AvroSchema.of(org.apache.pulsar.client.api.schema.proto.Test.TestMessageWrong.class)) + .newConsumer(AvroSchema.of + (SchemaDefinition.builder(). + withPojo(org.apache.pulsar.client.api.schema.proto.Test.TestMessageWrong.class).build())) .topic("persistent://my-property/use/my-ns/my-topic1") .subscriptionName("my-subscriber-name") .subscribe(); @@ -286,7 +289,8 @@ public void testAvroProducerAndConsumer() throws Exception { log.info("-- Starting {} test --", methodName); AvroSchema avroSchema = - AvroSchema.of(AvroEncodedPojo.class); + AvroSchema.of(SchemaDefinition.builder(). + withPojo(AvroEncodedPojo.class).build()); Consumer consumer = pulsarClient .newConsumer(avroSchema) @@ -355,7 +359,8 @@ public void testAvroConsumerWithWrongPrestoredSchema() throws Exception { ).get(); Consumer consumer = pulsarClient - .newConsumer(AvroSchema.of(AvroEncodedPojo.class)) + .newConsumer(AvroSchema.of(SchemaDefinition.builder(). + withPojo(AvroEncodedPojo.class).build())) .topic("persistent://my-property/use/my-ns/my-topic1") .subscriptionName("my-subscriber-name") .subscribe(); @@ -454,7 +459,8 @@ public void testAvroProducerAndAutoSchemaConsumer() throws Exception { log.info("-- Starting {} test --", methodName); AvroSchema avroSchema = - AvroSchema.of(AvroEncodedPojo.class); + AvroSchema.of(SchemaDefinition.builder(). + withPojo(AvroEncodedPojo.class).build()); Producer producer = pulsarClient .newProducer(avroSchema) @@ -502,7 +508,8 @@ public void testAvroProducerAndAutoSchemaReader() throws Exception { log.info("-- Starting {} test --", methodName); AvroSchema avroSchema = - AvroSchema.of(AvroEncodedPojo.class); + AvroSchema.of(SchemaDefinition.builder(). + withPojo(AvroEncodedPojo.class).build()); Producer producer = pulsarClient .newProducer(avroSchema) @@ -548,7 +555,8 @@ public void testAutoBytesProducer() throws Exception { log.info("-- Starting {} test --", methodName); AvroSchema avroSchema = - AvroSchema.of(AvroEncodedPojo.class); + AvroSchema.of(SchemaDefinition.builder(). + withPojo(AvroEncodedPojo.class).build()); try (Producer producer = pulsarClient .newProducer(avroSchema) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java index e17fb0608df85..d0bff30bdd032 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java @@ -19,10 +19,10 @@ package org.apache.pulsar.client.api; import java.nio.ByteBuffer; -import java.util.Collections; -import java.util.Map; + import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.api.schema.GenericSchema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.internal.DefaultImplementation; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.SchemaInfo; @@ -169,58 +169,43 @@ static Schema PROTOBUF(Cla } /** - * Create a Avro schema type by extracting the fields of the specified class. - * - * @param clazz the POJO class to be used to extract the Avro schema - * @return a Schema instance - */ - static Schema AVRO(Class clazz) { - return DefaultImplementation.newAvroSchema(clazz); - } - - /** - * Create a Avro schema type using the provided avro schema definition. + * Create a Avro schema type by default configuration of the class * - * @param schemaDefinition avro schema definition + * @param pojo the POJO class to be used to extract the Avro schema * @return a Schema instance */ - static Schema AVRO(String schemaDefinition) { - return AVRO(schemaDefinition, Collections.emptyMap()); + static Schema AVRO(Class pojo) { + return DefaultImplementation.newAvroSchema(SchemaDefinition.builder().withPojo(pojo).build()); } /** - * Create a Avro schema type using the provided avro schema definition. + * Create a Avro schema type with schema definition * - * @param schemaDefinition avro schema definition - * @param properties pulsar schema properties + * @param schemaDefinition the definition of the schema * @return a Schema instance */ - static Schema AVRO(String schemaDefinition, Map properties) { - return DefaultImplementation.newAvroSchema(schemaDefinition, properties); + static Schema AVRO(SchemaDefinition schemaDefinition) { + return DefaultImplementation.newAvroSchema(schemaDefinition); } /** * Create a JSON schema type by extracting the fields of the specified class. * - * @param clazz the POJO class to be used to extract the JSON schema + * @param pojo the POJO class to be used to extract the JSON schema * @return a Schema instance */ - static Schema JSON(Class clazz) { - return DefaultImplementation.newJSONSchema(clazz); + static Schema JSON(Class pojo) { + return DefaultImplementation.newJSONSchema(SchemaDefinition.builder().withPojo(pojo).build()); } /** - * Create a JSON schema type by extracting the fields of the specified class. + * Create a JSON schema type with schema definition * - * @param clazz the POJO class to be used to extract the JSON schema - * @param schemaDefinition schema definition json string (using avro schema syntax) - * @param properties pulsar schema properties + * @param schemaDefinition the definition of the schema * @return a Schema instance */ - static Schema JSON(Class clazz, - String schemaDefinition, - Map properties) { - return DefaultImplementation.newJSONSchema(clazz, schemaDefinition, properties); + static Schema JSON(SchemaDefinition schemaDefinition) { + return DefaultImplementation.newJSONSchema(schemaDefinition); } /** diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java new file mode 100644 index 0000000000000..daf90dfd8ea29 --- /dev/null +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java @@ -0,0 +1,64 @@ +/** + * 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.schema; + +import org.apache.pulsar.client.internal.DefaultImplementation; + +import java.util.Map; + + +public interface SchemaDefinition { + + /** + * Get a new builder instance that can used to configure and build a {@link SchemaDefinition} instance. + * + * @return the {@link SchemaDefinition} + */ + static SchemaDefinitionBuilder builder() { + return DefaultImplementation.newSchemaDefinitionBuilder(); + } + + /** + * get schema whether always allow null or not + * + * @return schema always null or not + */ + public boolean getAlwaysAllowNull(); + + /** + * Get schema class + * + * @return schema class + */ + public Map getProperties(); + + /** + * Get json schema definition + * + * @return schema class + */ + public String getJsonDef(); + + /** + * Get pojo schema definition + * + * @return pojo schema + */ + public Class getPojo(); +} diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinitionBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinitionBuilder.java new file mode 100644 index 0000000000000..77bb363cc713c --- /dev/null +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinitionBuilder.java @@ -0,0 +1,81 @@ +/** + * 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.schema; + + + +import java.util.Map; + +/** + * Builder to build schema definition {@link SchemaDefinition}. + */ +public interface SchemaDefinitionBuilder { + + /** + * Set schema whether always allow null or not + * + * @param alwaysAllowNull definition null or not + * @return schema definition builder + */ + SchemaDefinitionBuilder withAlwaysAllowNull(boolean alwaysAllowNull); + + /** + * Set schema info properties + * + * @param properties schema info properties + * @return schema definition builder + */ + SchemaDefinitionBuilder withProperties(Map properties); + + /** + * Set schema info properties + * + * @param key property key + * @param value property value + * + * @return record schema definition + */ + SchemaDefinitionBuilder addProperty(String key, String value); + + /** + * Set schema of pojo definition + * + * @param pojo pojo schema definition + * + * @return record schema definition + */ + SchemaDefinitionBuilder withPojo(Class pojo); + + /** + * Set schema of json definition + * + * @param jsonDefinition json schema definition + * + * @return record schema definition + */ + SchemaDefinitionBuilder withJsonDef(String jsonDefinition); + + /** + * Build the schema definition. + * + * @return the schema definition. + */ + SchemaDefinition build(); + +} diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java index abd368ede3ed2..44dbc10f22643 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java @@ -37,9 +37,7 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException; -import org.apache.pulsar.client.api.schema.GenericRecord; -import org.apache.pulsar.client.api.schema.GenericSchema; -import org.apache.pulsar.client.api.schema.RecordSchemaBuilder; +import org.apache.pulsar.client.api.schema.*; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; @@ -71,6 +69,13 @@ public class DefaultImplementation { private static final Constructor AUTHENTICATION_TLS_String_String = getConstructor( "org.apache.pulsar.client.impl.auth.AuthenticationTls", String.class, String.class); + private static final Constructor SCHEMA_DEFINITION_BUILDER_CONSTRUCTOR = getConstructor( + "org.apache.pulsar.client.impl.schema.SchemaDefinitionBuilderImpl"); + + public static SchemaDefinitionBuilder newSchemaDefinitionBuilder() { + return catchExceptions(() -> (SchemaDefinitionBuilder)SCHEMA_DEFINITION_BUILDER_CONSTRUCTOR.newInstance()); + } + public static ClientBuilder newClientBuilder() { return catchExceptions(() -> CLIENT_BUILDER_IMPL.newInstance()); } @@ -182,16 +187,10 @@ public static Schema newDoubleSchema() { .newInstance()); } - public static Schema newAvroSchema(Class clazz) { + public static Schema newAvroSchema(SchemaDefinition schemaDefinition) { return catchExceptions( - () -> (Schema) getStaticMethod("org.apache.pulsar.client.impl.schema.AvroSchema", "of", Class.class) - .invoke(null, clazz)); - } - - public static Schema newAvroSchema(String schemaDefinition, Map properties) { - return catchExceptions( - () -> (Schema) getStaticMethod("org.apache.pulsar.client.impl.schema.AvroSchema", "of", String.class, Map.class) - .invoke(null, schemaDefinition, properties)); + () -> (Schema) getStaticMethod("org.apache.pulsar.client.impl.schema.AvroSchema", "of", SchemaDefinition.class) + .invoke(null,schemaDefinition)); } public static Schema newProtobufSchema(Class clazz) { @@ -200,18 +199,10 @@ public static Schema newPr .invoke(null, clazz)); } - public static Schema newJSONSchema(Class clazz) { - return catchExceptions( - () -> (Schema) getStaticMethod("org.apache.pulsar.client.impl.schema.JSONSchema", "of", Class.class) - .invoke(null, clazz)); - } - - public static Schema newJSONSchema(Class clazz, - String schemaDefinition, - Map properties) { + public static Schema newJSONSchema(SchemaDefinition schemaDefinition) { return catchExceptions( - () -> (Schema) getStaticMethod("org.apache.pulsar.client.impl.schema.JSONSchema", "of", Class.class, String.class, Map.class) - .invoke(null, clazz, schemaDefinition, properties)); + () -> (Schema) getStaticMethod("org.apache.pulsar.client.impl.schema.JSONSchema", "of", SchemaDefinition.class) + .invoke(null, schemaDefinition)); } public static Schema newAutoConsumeSchema() { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java index c9726aa5bff69..a00112cb57a69 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java @@ -26,11 +26,12 @@ import org.apache.avro.reflect.ReflectDatumReader; import org.apache.avro.reflect.ReflectDatumWriter; import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.api.schema.SchemaDefinition; +import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.util.Collections; import java.util.Map; /** @@ -48,12 +49,11 @@ public class AvroSchema extends StructSchema { new ThreadLocal<>(); private AvroSchema(org.apache.avro.Schema schema, - Map properties) { + SchemaDefinition schemaDefinition) { super( SchemaType.AVRO, schema, - properties); - + schemaDefinition.getProperties()); this.byteArrayOutputStream = new ByteArrayOutputStream(); this.encoder = EncoderFactory.get().binaryEncoder(this.byteArrayOutputStream, this.encoder); this.datumWriter = new ReflectDatumWriter<>(this.schema); @@ -87,23 +87,23 @@ public T decode(byte[] bytes) { } } - public static AvroSchema of(Class pojo) { - return new AvroSchema<>(createAvroSchema(pojo), Collections.emptyMap()); + @Override + public SchemaInfo getSchemaInfo() { + return this.schemaInfo; } - public static AvroSchema of(Class pojo, Map properties) { - return new AvroSchema<>(createAvroSchema(pojo), properties); + public static AvroSchema of(SchemaDefinition schemaDefinition) { + return schemaDefinition.getJsonDef() == null ? + new AvroSchema<>(createAvroSchema(schemaDefinition), schemaDefinition) : new AvroSchema<>(parseAvroSchema(schemaDefinition.getJsonDef()), schemaDefinition); } - /** - * Create an Avro schema based on provided schema definition. - * - * @param schemaDefinition avro schema definition - * @param properties schema properties - * @return avro schema instance - */ - public static AvroSchema of(String schemaDefinition, Map properties) { - return new AvroSchema<>(parseAvroSchema(schemaDefinition), properties); + public static AvroSchema of(Class pojo) { + return AvroSchema.of(SchemaDefinition.builder().withPojo(pojo).build()); + } + + public static AvroSchema of(Class pojo, Map properties) { + SchemaDefinition schemaDefinition = SchemaDefinition.builder().withPojo(pojo).withProperties(properties).build(); + return new AvroSchema<>(createAvroSchema(schemaDefinition), schemaDefinition); } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java index b915ff20443ed..629b7697b03a0 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/JSONSchema.java @@ -26,11 +26,11 @@ import com.fasterxml.jackson.module.jsonSchema.JsonSchemaGenerator; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; import java.io.IOException; -import java.util.Collections; import java.util.Map; /** @@ -38,7 +38,6 @@ */ @Slf4j public class JSONSchema extends StructSchema { - // Cannot use org.apache.pulsar.common.util.ObjectMapperFactory.getThreadLocal() because it does not // return shaded version of object mapper private static final ThreadLocal JSON_MAPPER = ThreadLocal.withInitial(() -> { @@ -51,14 +50,13 @@ public class JSONSchema extends StructSchema { private final Class pojo; private final ObjectMapper objectMapper; - private JSONSchema(Class pojo, - org.apache.avro.Schema schema, - Map properties) { + private JSONSchema(org.apache.avro.Schema schema, + SchemaDefinition schemaDefinition) { super( SchemaType.JSON, schema, - properties); - this.pojo = pojo; + schemaDefinition.getProperties()); + this.pojo = schemaDefinition.getPojo(); this.objectMapper = JSON_MAPPER.get(); } @@ -89,6 +87,7 @@ public SchemaInfo getSchemaInfo() { * Implemented for backwards compatibility reasons * since the original schema generated by JSONSchema was based off the json schema standard * since then we have standardized on Avro + * * @return */ public SchemaInfo getBackwardsCompatibleJsonSchemaInfo() { @@ -108,25 +107,18 @@ public SchemaInfo getBackwardsCompatibleJsonSchemaInfo() { return backwardsCompatibleSchemaInfo; } + public static JSONSchema of(SchemaDefinition schemaDefinition) { + String jsonDef = schemaDefinition.getJsonDef(); + return jsonDef == null ? new JSONSchema<>(createAvroSchema(schemaDefinition), schemaDefinition) : + new JSONSchema<>(parseAvroSchema(jsonDef), schemaDefinition); + } + public static JSONSchema of(Class pojo) { - return new JSONSchema<>(pojo, createAvroSchema(pojo), Collections.emptyMap()); + return JSONSchema.of(SchemaDefinition.builder().withPojo(pojo).build()); } public static JSONSchema of(Class pojo, Map properties) { - return new JSONSchema<>(pojo, createAvroSchema(pojo), properties); + return JSONSchema.of(SchemaDefinition.builder().withPojo(pojo).withProperties(properties).build()); } - /** - * Create an json schema based on provided schema definition. - * - * @param pojo pojo class - * @param schemaDefinition avro schema definition - * @param properties schema properties - * @return avro schema instance - */ - public static JSONSchema of(Class pojo, - String schemaDefinition, - Map properties) { - return new JSONSchema<>(pojo, parseAvroSchema(schemaDefinition), properties); - } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionBuilderImpl.java new file mode 100644 index 0000000000000..2db6cf4c9610f --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionBuilderImpl.java @@ -0,0 +1,95 @@ +/** + * 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.impl.schema; + +import org.apache.pulsar.client.api.schema.SchemaDefinition; +import org.apache.pulsar.client.api.schema.SchemaDefinitionBuilder; + +import java.util.HashMap; +import java.util.Map; + +/** + * Builder to build {@link org.apache.pulsar.client.api.schema.GenericRecord}. + */ +public class SchemaDefinitionBuilderImpl implements SchemaDefinitionBuilder { + + public static final String ALWAYS_ALLOW_NULL = "__alwaysAllowNull"; + + /** + * the schema definition class + */ + private Class clazz; + /** + * The flag of schema type always allow null + * + * If it's true, will make all of the pojo field generate schema + * define default can be null,false default can't be null, but it's + * false you can define the field by yourself by the annotation@Nullable + * + */ + private boolean alwaysAllowNull = true; + + /** + * The schema info properties + */ + private Map properties = new HashMap<>(); + + /** + * The json schema definition + */ + private String jsonDef; + + @Override + public SchemaDefinitionBuilder withAlwaysAllowNull(boolean alwaysAllowNull) { + this.alwaysAllowNull = alwaysAllowNull; + return this; + } + + @Override + public SchemaDefinitionBuilder addProperty(String key, String value) { + this.properties.put(key, value); + return this; + } + + @Override + public SchemaDefinitionBuilder withPojo(Class clazz) { + this.clazz = clazz; + return this; + } + + @Override + public SchemaDefinitionBuilder withJsonDef(String jsonDef) { + this.jsonDef = jsonDef; + return this; + } + + + @Override + public SchemaDefinitionBuilder withProperties(Map properties) { + this.properties = properties; + return this; + } + + @Override + public SchemaDefinition build() { + properties.put(ALWAYS_ALLOW_NULL, this.alwaysAllowNull ? "true" : "false"); + return new SchemaDefinitionImpl(clazz, jsonDef, alwaysAllowNull, properties); + + } +} diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionImpl.java new file mode 100644 index 0000000000000..04f1a2441a89d --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/SchemaDefinitionImpl.java @@ -0,0 +1,98 @@ +/** + * 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.impl.schema; + + +import org.apache.pulsar.client.api.schema.SchemaDefinition; + +import java.util.HashMap; +import java.util.Map; + +/** + * A json schema definition + * {@link org.apache.pulsar.client.api.schema.SchemaDefinition} for the json schema definition. + */ +public class SchemaDefinitionImpl implements SchemaDefinition{ + + /** + * the schema definition class + */ + private Class pojo; + /** + * The flag of schema type always allow null + * + * If it's true, will make all of the pojo field generate schema + * define default can be null,false default can't be null, but it's + * false you can define the field by yourself by the annotation@Nullable + * + */ + private boolean alwaysAllowNull; + + private Map properties; + + private String jsonDef; + + public SchemaDefinitionImpl(Class pojo, String jsonDef, boolean alwaysAllowNull, Map properties) { + this.alwaysAllowNull = alwaysAllowNull; + this.properties = properties; + this.jsonDef = jsonDef; + this.pojo = pojo; + } + /** + * get schema whether always allow null or not + * + * @return schema always null or not + */ + public boolean getAlwaysAllowNull() { + + return alwaysAllowNull; + } + + /** + * Get json schema definition + * + * @return schema class + */ + public String getJsonDef() { + + return jsonDef; + } + /** + * Get pojo schema definition + * + * @return pojo class + */ + @Override + public Class getPojo() { + return pojo; + } + + /** + * Get schema class + * + * @return schema class + */ + public Map getProperties() { + + return properties; + } + + + +} diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java index 33ce9de97ce6f..31156d4d49925 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java @@ -24,6 +24,7 @@ import org.apache.avro.Schema.Parser; import org.apache.avro.reflect.ReflectData; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; @@ -62,13 +63,14 @@ public SchemaInfo getSchemaInfo() { return this.schemaInfo; } - protected static org.apache.avro.Schema createAvroSchema(Class pojo) { - return ReflectData.AllowNull.get().getSchema(pojo); + protected static org.apache.avro.Schema createAvroSchema(SchemaDefinition schemaDefinition) { + Class pojo = schemaDefinition.getPojo(); + return schemaDefinition.getAlwaysAllowNull() ? ReflectData.AllowNull.get().getSchema(pojo) : ReflectData.get().getSchema(pojo); } - protected static org.apache.avro.Schema parseAvroSchema(String definition) { + protected static org.apache.avro.Schema parseAvroSchema(String jsonDef) { Parser parser = new Parser(); - return parser.parse(definition); + return parser.parse(jsonDef); } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java index 203c22622c745..bbd753bbe9b76 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java @@ -20,29 +20,34 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.FOO_FIELDS; -import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.SCHEMA_JSON; +import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.SCHEMA_AVRO_NOT_ALLOW_NULL; +import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.SCHEMA_AVRO_ALLOW_NULL; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.fail; import java.util.Arrays; -import java.util.Collections; + import lombok.Data; import lombok.extern.slf4j.Slf4j; + import org.apache.avro.Schema; +import org.apache.pulsar.client.api.SchemaSerializationException; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.avro.SchemaValidationException; import org.apache.avro.SchemaValidator; import org.apache.avro.SchemaValidatorBuilder; import org.apache.avro.reflect.AvroDefault; import org.apache.avro.reflect.Nullable; import org.apache.avro.reflect.ReflectData; -import org.apache.pulsar.client.api.SchemaSerializationException; + import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Bar; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Foo; import org.apache.pulsar.common.schema.SchemaType; import org.testng.Assert; import org.testng.annotations.Test; + @Slf4j public class AvroSchemaTest { @@ -88,7 +93,7 @@ public void testSchemaDefinition() throws SchemaValidationException { // expected } - AvroSchema schema3 = AvroSchema.of(schemaDef1, Collections.emptyMap()); + AvroSchema schema3 = AvroSchema.of(SchemaDefinition.builder().withJsonDef(schemaDef1).build()); String schemaDef3 = new String(schema3.getSchemaInfo().getSchema(), UTF_8); assertEquals(schemaDef1, schemaDef3); assertNotEquals(schemaDef2, schemaDef3); @@ -108,12 +113,12 @@ public void testSchemaDefinition() throws SchemaValidationException { } @Test - public void testSchema() { - AvroSchema avroSchema = AvroSchema.of(Foo.class); + public void testNotAllowNullSchema() { + AvroSchema avroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); assertEquals(avroSchema.getSchemaInfo().getType(), SchemaType.AVRO); Schema.Parser parser = new Schema.Parser(); String schemaJson = new String(avroSchema.getSchemaInfo().getSchema()); - assertEquals(schemaJson, SCHEMA_JSON); + assertEquals(schemaJson, SCHEMA_AVRO_NOT_ALLOW_NULL); Schema schema = parser.parse(schemaJson); for (String fieldName : FOO_FIELDS) { @@ -123,12 +128,66 @@ public void testSchema() { if (field.name().equals("field4")) { Assert.assertNotNull(field.schema().getTypes().get(1).getField("field1")); } + if (field.name().equals("fieldUnableNull")) { + Assert.assertNotNull(field.schema().getType()); + } } } @Test - public void testEncodeAndDecode() { - AvroSchema avroSchema = AvroSchema.of(Foo.class, null); + public void testAllowNullSchema() { + AvroSchema avroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + assertEquals(avroSchema.getSchemaInfo().getType(), SchemaType.AVRO); + Schema.Parser parser = new Schema.Parser(); + String schemaJson = new String(avroSchema.getSchemaInfo().getSchema()); + assertEquals(schemaJson, SCHEMA_AVRO_ALLOW_NULL); + Schema schema = parser.parse(schemaJson); + + for (String fieldName : FOO_FIELDS) { + Schema.Field field = schema.getField(fieldName); + Assert.assertNotNull(field); + + if (field.name().equals("field4")) { + Assert.assertNotNull(field.schema().getTypes().get(1).getField("field1")); + } + if (field.name().equals("fieldUnableNull")) { + Assert.assertNotNull(field.schema().getType()); + } + } + } + + @Test + public void testNotAllowNullEncodeAndDecode() { + AvroSchema avroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); + + Foo foo1 = new Foo(); + foo1.setField1("foo1"); + foo1.setField2("bar1"); + foo1.setField4(new Bar()); + foo1.setFieldUnableNull("notNull"); + + Foo foo2 = new Foo(); + foo2.setField1("foo2"); + foo2.setField2("bar2"); + + byte[] bytes1 = avroSchema.encode(foo1); + Foo object1 = avroSchema.decode(bytes1); + Assert.assertTrue(bytes1.length > 0); + assertEquals(object1, foo1); + + try { + + avroSchema.encode(foo2); + + } catch (Exception e) { + Assert.assertTrue(e instanceof SchemaSerializationException); + } + + } + + @Test + public void testAllowNullEncodeAndDecode() { + AvroSchema avroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); Foo foo1 = new Foo(); foo1.setField1("foo1"); @@ -150,6 +209,8 @@ public void testEncodeAndDecode() { assertEquals(object1, foo1); assertEquals(object2, foo2); + } + } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/JSONSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/JSONSchemaTest.java index 9184faa45910f..5efb82bd013bb 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/JSONSchemaTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/JSONSchemaTest.java @@ -24,7 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.avro.Schema; import org.apache.pulsar.client.api.SchemaSerializationException; -import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Bar; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.DerivedFoo; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Foo; @@ -35,18 +35,20 @@ import org.testng.annotations.Test; import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.FOO_FIELDS; -import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.SCHEMA_JSON; +import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.SCHEMA_JSON_NOT_ALLOW_NULL; +import static org.apache.pulsar.client.impl.schema.SchemaTestUtils.SCHEMA_JSON_ALLOW_NULL; +import static org.testng.Assert.assertEquals; @Slf4j public class JSONSchemaTest { @Test - public void testSchema() { - JSONSchema jsonSchema = JSONSchema.of(Foo.class); + public void testNotAllowNullSchema() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); Assert.assertEquals(jsonSchema.getSchemaInfo().getType(), SchemaType.JSON); Schema.Parser parser = new Schema.Parser(); String schemaJson = new String(jsonSchema.getSchemaInfo().getSchema()); - Assert.assertEquals(schemaJson, SCHEMA_JSON); + Assert.assertEquals(schemaJson, SCHEMA_JSON_NOT_ALLOW_NULL); Schema schema = parser.parse(schemaJson); for (String fieldName : FOO_FIELDS) { @@ -56,12 +58,37 @@ public void testSchema() { if (field.name().equals("field4")) { Assert.assertNotNull(field.schema().getTypes().get(1).getField("field1")); } + if (field.name().equals("fieldUnableNull")) { + Assert.assertNotNull(field.schema().getType()); + } } } @Test - public void testEncodeAndDecode() { - JSONSchema jsonSchema = JSONSchema.of(Foo.class, null); + public void testAllowNullSchema() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + Assert.assertEquals(jsonSchema.getSchemaInfo().getType(), SchemaType.JSON); + Schema.Parser parser = new Schema.Parser(); + String schemaJson = new String(jsonSchema.getSchemaInfo().getSchema()); + Assert.assertEquals(schemaJson, SCHEMA_JSON_ALLOW_NULL); + Schema schema = parser.parse(schemaJson); + + for (String fieldName : FOO_FIELDS) { + Schema.Field field = schema.getField(fieldName); + Assert.assertNotNull(field); + + if (field.name().equals("field4")) { + Assert.assertNotNull(field.schema().getTypes().get(1).getField("field1")); + } + if (field.name().equals("fieldUnableNull")) { + Assert.assertNotNull(field.schema().getType()); + } + } + } + + @Test + public void testAllowNullEncodeAndDecode() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); Bar bar = new Bar(); bar.setField1(true); @@ -90,9 +117,65 @@ public void testEncodeAndDecode() { } @Test - public void testNestedClasses() { - JSONSchema jsonSchema = JSONSchema.of(NestedBar.class, null); - JSONSchema listJsonSchema = JSONSchema.of(NestedBarList.class, null); + public void testNotAllowNullEncodeAndDecode() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); + + Foo foo1 = new Foo(); + foo1.setField1("foo1"); + foo1.setField2("bar1"); + foo1.setField4(new Bar()); + foo1.setFieldUnableNull("notNull"); + + Foo foo2 = new Foo(); + foo2.setField1("foo2"); + foo2.setField2("bar2"); + + byte[] bytes1 = jsonSchema.encode(foo1); + Foo object1 = jsonSchema.decode(bytes1); + Assert.assertTrue(bytes1.length > 0); + assertEquals(object1, foo1); + + try { + + jsonSchema.encode(foo2); + + } catch (Exception e) { + Assert.assertTrue(e instanceof SchemaSerializationException); + } + + } + + @Test + public void testAllowNullNestedClasses() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(NestedBar.class).build()); + JSONSchema listJsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(NestedBarList.class).build()); + + Bar bar = new Bar(); + bar.setField1(true); + + NestedBar nested = new NestedBar(); + nested.setField1(true); + nested.setNested(bar); + + byte[] bytes = jsonSchema.encode(nested); + Assert.assertTrue(bytes.length > 0); + Assert.assertEquals(jsonSchema.decode(bytes), nested); + + List list = Collections.singletonList(bar); + NestedBarList nestedList = new NestedBarList(); + nestedList.setField1(true); + nestedList.setList(list); + + bytes = listJsonSchema.encode(nestedList); + Assert.assertTrue(bytes.length > 0); + + Assert.assertEquals(listJsonSchema.decode(bytes), nestedList); + } + + @Test + public void testNotAllowNullNestedClasses() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(NestedBar.class).withAlwaysAllowNull(false).build()); + JSONSchema listJsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(NestedBarList.class).withAlwaysAllowNull(false).build()); Bar bar = new Bar(); bar.setField1(true); @@ -117,7 +200,59 @@ public void testNestedClasses() { } @Test - public void testCorrectPolymorphism() { + public void testNotAllowNullCorrectPolymorphism() { + Bar bar = new Bar(); + bar.setField1(true); + + DerivedFoo derivedFoo = new DerivedFoo(); + derivedFoo.setField1("foo1"); + derivedFoo.setField2("bar2"); + derivedFoo.setField3(4); + derivedFoo.setField4(bar); + derivedFoo.setField5("derived1"); + derivedFoo.setField6(2); + + Foo foo = new Foo(); + foo.setField1("foo1"); + foo.setField2("bar2"); + foo.setField3(4); + foo.setField4(bar); + + SchemaTestUtils.DerivedDerivedFoo derivedDerivedFoo = new SchemaTestUtils.DerivedDerivedFoo(); + derivedDerivedFoo.setField1("foo1"); + derivedDerivedFoo.setField2("bar2"); + derivedDerivedFoo.setField3(4); + derivedDerivedFoo.setField4(bar); + derivedDerivedFoo.setField5("derived1"); + derivedDerivedFoo.setField6(2); + derivedDerivedFoo.setFoo2(foo); + derivedDerivedFoo.setDerivedFoo(derivedFoo); + + // schema for base class + JSONSchema baseJsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + Assert.assertEquals(baseJsonSchema.decode(baseJsonSchema.encode(foo)), foo); + Assert.assertEquals(baseJsonSchema.decode(baseJsonSchema.encode(derivedFoo)), foo); + Assert.assertEquals(baseJsonSchema.decode(baseJsonSchema.encode(derivedDerivedFoo)), foo); + + // schema for derived class + JSONSchema derivedJsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(DerivedFoo.class).build()); + Assert.assertEquals(derivedJsonSchema.decode(derivedJsonSchema.encode(derivedFoo)), derivedFoo); + Assert.assertEquals(derivedJsonSchema.decode(derivedJsonSchema.encode(derivedDerivedFoo)), derivedFoo); + + //schema for derived derived class + JSONSchema derivedDerivedJsonSchema + = JSONSchema.of(SchemaDefinition.builder().withPojo(SchemaTestUtils.DerivedDerivedFoo.class).build()); + Assert.assertEquals(derivedDerivedJsonSchema.decode(derivedDerivedJsonSchema.encode(derivedDerivedFoo)), derivedDerivedFoo); + } + + @Test(expectedExceptions = SchemaSerializationException.class) + public void testAllowNullDecodeWithInvalidContent() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + jsonSchema.decode(new byte[0]); + } + + @Test + public void testAllowNullCorrectPolymorphism() { Bar bar = new Bar(); bar.setField1(true); @@ -146,25 +281,25 @@ public void testCorrectPolymorphism() { derivedDerivedFoo.setDerivedFoo(derivedFoo); // schema for base class - JSONSchema baseJsonSchema = JSONSchema.of(Foo.class); + JSONSchema baseJsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); Assert.assertEquals(baseJsonSchema.decode(baseJsonSchema.encode(foo)), foo); Assert.assertEquals(baseJsonSchema.decode(baseJsonSchema.encode(derivedFoo)), foo); Assert.assertEquals(baseJsonSchema.decode(baseJsonSchema.encode(derivedDerivedFoo)), foo); // schema for derived class - JSONSchema derivedJsonSchema = JSONSchema.of(DerivedFoo.class); + JSONSchema derivedJsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(DerivedFoo.class).withAlwaysAllowNull(false).build()); Assert.assertEquals(derivedJsonSchema.decode(derivedJsonSchema.encode(derivedFoo)), derivedFoo); Assert.assertEquals(derivedJsonSchema.decode(derivedJsonSchema.encode(derivedDerivedFoo)), derivedFoo); //schema for derived derived class JSONSchema derivedDerivedJsonSchema - = JSONSchema.of(SchemaTestUtils.DerivedDerivedFoo.class); + = JSONSchema.of(SchemaDefinition.builder().withPojo(SchemaTestUtils.DerivedDerivedFoo.class).withAlwaysAllowNull(false).build()); Assert.assertEquals(derivedDerivedJsonSchema.decode(derivedDerivedJsonSchema.encode(derivedDerivedFoo)), derivedDerivedFoo); } @Test(expectedExceptions = SchemaSerializationException.class) - public void testDecodeWithInvalidContent() { - JSONSchema jsonSchema = JSONSchema.of(Foo.class); + public void testNotAllowNullDecodeWithInvalidContent() { + JSONSchema jsonSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); jsonSchema.decode(new byte[0]); } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java index 2963d622f737f..a32e31ddcadca 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/KeyValueSchemaTest.java @@ -22,9 +22,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.impl.schema.AvroSchema; -import org.apache.pulsar.client.impl.schema.JSONSchema; -import org.apache.pulsar.client.impl.schema.KeyValueSchema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Bar; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Color; import org.apache.pulsar.client.impl.schema.SchemaTestUtils.Foo; @@ -37,9 +35,9 @@ public class KeyValueSchemaTest { @Test - public void testAvroSchemaCreate() { - AvroSchema fooSchema = AvroSchema.of(Foo.class); - AvroSchema barSchema = AvroSchema.of(Bar.class); + public void testAllowNullAvroSchemaCreate() { + AvroSchema fooSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + AvroSchema barSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Bar.class).build()); Schema> keyValueSchema1 = Schema.KeyValue(fooSchema, barSchema); Schema> keyValueSchema2 = Schema.KeyValue(Foo.class, Bar.class, SchemaType.AVRO); @@ -47,13 +45,39 @@ public void testAvroSchemaCreate() { assertEquals(keyValueSchema1.getSchemaInfo().getType(), SchemaType.KEY_VALUE); assertEquals(keyValueSchema2.getSchemaInfo().getType(), SchemaType.KEY_VALUE); - assertEquals(((KeyValueSchema)keyValueSchema1).getKeySchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema1).getKeySchema().getSchemaInfo().getType(), + SchemaType.AVRO); + assertEquals(((KeyValueSchema) keyValueSchema1).getValueSchema().getSchemaInfo().getType(), + SchemaType.AVRO); + assertEquals(((KeyValueSchema) keyValueSchema2).getKeySchema().getSchemaInfo().getType(), + SchemaType.AVRO); + assertEquals(((KeyValueSchema) keyValueSchema2).getValueSchema().getSchemaInfo().getType(), + SchemaType.AVRO); + + String schemaInfo1 = new String(keyValueSchema1.getSchemaInfo().getSchema()); + String schemaInfo2 = new String(keyValueSchema2.getSchemaInfo().getSchema()); + assertEquals(schemaInfo1, schemaInfo2); + } + + @Test + public void testNotAllowNullAvroSchemaCreate() { + AvroSchema fooSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); + AvroSchema barSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build()); + + Schema> keyValueSchema1 = Schema.KeyValue(fooSchema, barSchema); + Schema> keyValueSchema2 = Schema.KeyValue(AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()), + AvroSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build())); + + assertEquals(keyValueSchema1.getSchemaInfo().getType(), SchemaType.KEY_VALUE); + assertEquals(keyValueSchema2.getSchemaInfo().getType(), SchemaType.KEY_VALUE); + + assertEquals(((KeyValueSchema) keyValueSchema1).getKeySchema().getSchemaInfo().getType(), SchemaType.AVRO); - assertEquals(((KeyValueSchema)keyValueSchema1).getValueSchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema1).getValueSchema().getSchemaInfo().getType(), SchemaType.AVRO); - assertEquals(((KeyValueSchema)keyValueSchema2).getKeySchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema2).getKeySchema().getSchemaInfo().getType(), SchemaType.AVRO); - assertEquals(((KeyValueSchema)keyValueSchema2).getValueSchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema2).getValueSchema().getSchemaInfo().getType(), SchemaType.AVRO); String schemaInfo1 = new String(keyValueSchema1.getSchemaInfo().getSchema()); @@ -62,9 +86,9 @@ public void testAvroSchemaCreate() { } @Test - public void testJsonSchemaCreate() { - JSONSchema fooSchema = JSONSchema.of(Foo.class); - JSONSchema barSchema = JSONSchema.of(Bar.class); + public void testAllowNullJsonSchemaCreate() { + JSONSchema fooSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + JSONSchema barSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Bar.class).build()); Schema> keyValueSchema1 = Schema.KeyValue(fooSchema, barSchema); Schema> keyValueSchema2 = Schema.KeyValue(Foo.class, Bar.class, SchemaType.JSON); @@ -74,17 +98,53 @@ public void testJsonSchemaCreate() { assertEquals(keyValueSchema2.getSchemaInfo().getType(), SchemaType.KEY_VALUE); assertEquals(keyValueSchema3.getSchemaInfo().getType(), SchemaType.KEY_VALUE); - assertEquals(((KeyValueSchema)keyValueSchema1).getKeySchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema1).getKeySchema().getSchemaInfo().getType(), + SchemaType.JSON); + assertEquals(((KeyValueSchema) keyValueSchema1).getValueSchema().getSchemaInfo().getType(), + SchemaType.JSON); + assertEquals(((KeyValueSchema) keyValueSchema2).getKeySchema().getSchemaInfo().getType(), + SchemaType.JSON); + assertEquals(((KeyValueSchema) keyValueSchema2).getValueSchema().getSchemaInfo().getType(), + SchemaType.JSON); + assertEquals(((KeyValueSchema) keyValueSchema3).getKeySchema().getSchemaInfo().getType(), + SchemaType.JSON); + assertEquals(((KeyValueSchema) keyValueSchema3).getValueSchema().getSchemaInfo().getType(), + SchemaType.JSON); + + String schemaInfo1 = new String(keyValueSchema1.getSchemaInfo().getSchema()); + String schemaInfo2 = new String(keyValueSchema2.getSchemaInfo().getSchema()); + String schemaInfo3 = new String(keyValueSchema3.getSchemaInfo().getSchema()); + assertEquals(schemaInfo1, schemaInfo2); + assertEquals(schemaInfo1, schemaInfo3); + } + + @Test + public void testNotAllowNullJsonSchemaCreate() { + JSONSchema fooSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); + JSONSchema barSchema = JSONSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build()); + + Schema> keyValueSchema1 = Schema.KeyValue(fooSchema, barSchema); + Schema> keyValueSchema2 = Schema.KeyValue(JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()), + JSONSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build())); + + Schema> keyValueSchema3 = Schema.KeyValue(JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()), + JSONSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build())); + + assertEquals(keyValueSchema1.getSchemaInfo().getType(), SchemaType.KEY_VALUE); + assertEquals(keyValueSchema2.getSchemaInfo().getType(), SchemaType.KEY_VALUE); + assertEquals(keyValueSchema3.getSchemaInfo().getType(), SchemaType.KEY_VALUE); + + assertEquals(((KeyValueSchema) keyValueSchema1).getKeySchema().getSchemaInfo().getType(), SchemaType.JSON); - assertEquals(((KeyValueSchema)keyValueSchema1).getValueSchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema1).getValueSchema().getSchemaInfo().getType(), SchemaType.JSON); - assertEquals(((KeyValueSchema)keyValueSchema2).getKeySchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema2).getKeySchema().getSchemaInfo().getType(), SchemaType.JSON); - assertEquals(((KeyValueSchema)keyValueSchema2).getValueSchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema2).getValueSchema().getSchemaInfo().getType(), SchemaType.JSON); - assertEquals(((KeyValueSchema)keyValueSchema3).getKeySchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema3).getKeySchema().getSchemaInfo().getType(), SchemaType.JSON); - assertEquals(((KeyValueSchema)keyValueSchema3).getValueSchema().getSchemaInfo().getType(), + assertEquals(((KeyValueSchema) keyValueSchema3).getValueSchema().getSchemaInfo().getType(), SchemaType.JSON); String schemaInfo1 = new String(keyValueSchema1.getSchemaInfo().getSchema()); @@ -95,7 +155,7 @@ public void testJsonSchemaCreate() { } @Test - public void testSchemaEncodeAndDecode() { + public void testAllowNullSchemaEncodeAndDecode() { Schema keyValueSchema = Schema.KeyValue(Foo.class, Bar.class); Bar bar = new Bar(); @@ -111,7 +171,7 @@ public void testSchemaEncodeAndDecode() { byte[] encodeBytes = keyValueSchema.encode(new KeyValue(foo, bar)); Assert.assertTrue(encodeBytes.length > 0); - KeyValue keyValue = (KeyValue)keyValueSchema.decode(encodeBytes); + KeyValue keyValue = (KeyValue) keyValueSchema.decode(encodeBytes); Foo fooBack = keyValue.getKey(); Bar barBack = keyValue.getValue(); @@ -120,9 +180,64 @@ public void testSchemaEncodeAndDecode() { } @Test - public void testBytesSchemaEncodeAndDecode() { - AvroSchema fooAvroSchema = AvroSchema.of(Foo.class); - AvroSchema barAvroSchema = AvroSchema.of(Bar.class); + public void testNotAllowNullSchemaEncodeAndDecode() { + Schema keyValueSchema = Schema.KeyValue(JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()), + JSONSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build())); + + Bar bar = new Bar(); + bar.setField1(true); + + Foo foo = new Foo(); + foo.setField1("field1"); + foo.setField2("field2"); + foo.setField3(3); + foo.setField4(bar); + foo.setColor(Color.RED); + + byte[] encodeBytes = keyValueSchema.encode(new KeyValue(foo, bar)); + Assert.assertTrue(encodeBytes.length > 0); + + KeyValue keyValue = (KeyValue) keyValueSchema.decode(encodeBytes); + Foo fooBack = keyValue.getKey(); + Bar barBack = keyValue.getValue(); + + assertEquals(foo, fooBack); + assertEquals(bar, barBack); + } + + @Test + public void testAllowNullBytesSchemaEncodeAndDecode() { + AvroSchema fooAvroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); + AvroSchema barAvroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Bar.class).build()); + + Bar bar = new Bar(); + bar.setField1(true); + + Foo foo = new Foo(); + foo.setField1("field1"); + foo.setField2("field2"); + foo.setField3(3); + foo.setField4(bar); + foo.setColor(Color.RED); + foo.setFieldUnableNull("notNull"); + + byte[] fooBytes = fooAvroSchema.encode(foo); + byte[] barBytes = barAvroSchema.encode(bar); + + byte[] encodeBytes = Schema.KV_BYTES().encode(new KeyValue<>(fooBytes, barBytes)); + KeyValue decodeKV = Schema.KV_BYTES().decode(encodeBytes); + + Foo fooBack = fooAvroSchema.decode(decodeKV.getKey()); + Bar barBack = barAvroSchema.decode(decodeKV.getValue()); + + assertEquals(foo, fooBack); + assertEquals(bar, barBack); + } + + @Test + public void testNotAllowNullBytesSchemaEncodeAndDecode() { + AvroSchema fooAvroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).withAlwaysAllowNull(false).build()); + AvroSchema barAvroSchema = AvroSchema.of(SchemaDefinition.builder().withPojo(Bar.class).withAlwaysAllowNull(false).build()); Bar bar = new Bar(); bar.setField1(true); @@ -133,6 +248,7 @@ public void testBytesSchemaEncodeAndDecode() { foo.setField3(3); foo.setField4(bar); foo.setColor(Color.RED); + foo.setFieldUnableNull("notNull"); byte[] fooBytes = fooAvroSchema.encode(foo); byte[] barBytes = barAvroSchema.encode(bar); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaBuilderTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaBuilderTest.java index 4bc0c045b853c..456416f99966f 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaBuilderTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaBuilderTest.java @@ -21,6 +21,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.testng.Assert.assertEquals; +import org.apache.avro.reflect.Nullable; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.api.schema.GenericSchema; @@ -36,11 +37,17 @@ public class SchemaBuilderTest { private static class AllOptionalFields { + @Nullable private Integer intField; + @Nullable private Long longField; + @Nullable private String stringField; + @Nullable private Boolean boolField; + @Nullable private Float floatField; + @Nullable private Double doubleField; } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaTestUtils.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaTestUtils.java index 0081f0e457ec1..98d53675b5301 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaTestUtils.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaTestUtils.java @@ -23,6 +23,8 @@ import lombok.Data; import lombok.EqualsAndHashCode; import lombok.ToString; +import org.apache.avro.reflect.AvroDefault; +import org.apache.avro.reflect.Nullable; /** * Utils for testing avro. @@ -33,11 +35,17 @@ public class SchemaTestUtils { @ToString @EqualsAndHashCode public static class Foo { + @Nullable private String field1; + @Nullable private String field2; private int field3; + @Nullable private Bar field4; + @Nullable private Color color; + @AvroDefault("\"defaultValue\"") + private String fieldUnableNull; } @Data @@ -87,21 +95,35 @@ public static class DerivedDerivedFoo extends DerivedFoo { private Foo foo2; } - public static final String SCHEMA_JSON - = "{\"type\":\"record\",\"name\":\"Foo\",\"namespace\":\"org.apache.pulsar.client.impl.schema" + - ".SchemaTestUtils$\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"],\"default\":null}," + - "{\"name\":\"field2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"field3\"," + - "\"type\":\"int\"},{\"name\":\"field4\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Bar\"," + - "\"fields\":[{\"name\":\"field1\",\"type\":\"boolean\"}]}],\"default\":null},{\"name\":\"color\"," + - "\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"Color\",\"symbols\":[\"RED\",\"BLUE\"]}]," + - "\"default\":null}]}"; + public static final String SCHEMA_AVRO_NOT_ALLOW_NULL + = "{\"type\":\"record\",\"name\":\"Foo\",\"namespace\":\"org.apache.pulsar.client.impl.schema.SchemaTestUtils$\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"]," + + "\"default\":null},{\"name\":\"field2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"field3\",\"type\":\"int\"},{\"name\":\"field4\",\"type\":[\"null\",{\"type\":" + + "\"record\",\"name\":\"Bar\",\"fields\":[{\"name\":\"field1\",\"type\":\"boolean\"}]}],\"default\":null},{\"name\":\"color\",\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"Color\"," + + "\"symbols\":[\"RED\",\"BLUE\"]}],\"default\":null},{\"name\":\"fieldUnableNull\",\"type\":\"string\",\"default\":\"defaultValue\"}]}"; + + public static final String SCHEMA_AVRO_ALLOW_NULL = "{\"type\":\"record\",\"name\":\"Foo\",\"namespace\":\"org.apache.pulsar.client.impl.schema.SchemaTestUtils$\",\"fields\":[{\"name\":\"field1\"," + + "\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"field2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"field3\",\"type\":\"int\"},{\"name\":\"field4\",\"type\":[\"" + + "null\",{\"type\":\"record\",\"name\":\"Bar\",\"fields\":[{\"name\":\"field1\",\"type\":\"boolean\"}]}],\"default\":null},{\"name\":\"color\",\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"Color\"" + + ",\"symbols\":[\"RED\",\"BLUE\"]}],\"default\":null},{\"name\":\"fieldUnableNull\",\"type\":[\"null\",\"string\"],\"default\":\"defaultValue\"}]}"; + + public static final String SCHEMA_JSON_NOT_ALLOW_NULL + = "{\"type\":\"record\",\"name\":\"Foo\",\"namespace\":\"org.apache.pulsar.client.impl.schema.SchemaTestUtils$\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\"" + + ":\"field2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"field3\",\"type\":\"int\"},{\"name\":\"field4\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Bar\",\"fields\":[{\"name\":\"" + + "field1\",\"type\":\"boolean\"}]}],\"default\":null},{\"name\":\"color\",\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"Color\",\"symbols\":[\"RED\",\"BLUE\"]}],\"default\":null},{\"name\":\"fieldUnableNull\"," + + "\"type\":\"string\",\"default\":\"defaultValue\"}]}"; + public static final String SCHEMA_JSON_ALLOW_NULL + = "{\"type\":\"record\",\"name\":\"Foo\",\"namespace\":\"org.apache.pulsar.client.impl.schema.SchemaTestUtils$\",\"fields\":[{\"name\":\"field1\",\"type\":[\"null\",\"string\"],\"default\":null}," + + "{\"name\":\"field2\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"field3\",\"type\":\"int\"},{\"name\":\"field4\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Bar\",\"fields\":" + + "[{\"name\":\"field1\",\"type\":\"boolean\"}]}],\"default\":null},{\"name\":\"color\",\"type\":[\"null\",{\"type\":\"enum\",\"name\":\"Color\",\"symbols\":[\"RED\",\"BLUE\"]}],\"default\":null},{\"name\":" + + "\"fieldUnableNull\",\"type\":[\"null\",\"string\"],\"default\":\"defaultValue\"}]}"; public static String[] FOO_FIELDS = { "field1", "field2", "field3", "field4", - "color" + "color", + "fieldUnableNull" }; } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java index fc65d70859624..ed554cce8e9a5 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java @@ -77,7 +77,7 @@ public void testEncodeAndDecodeGenericRecord(Schema encodeSchema, Bar bar = new Bar(); bar.setField1(i % 2 == 0); foo.setField4(bar); - + foo.setFieldUnableNull("fieldUnableNull-1-" + i); byte[] data = encodeSchema.encode(foo); log.info("Decoding : {}", new String(data, UTF_8)); @@ -93,6 +93,8 @@ public void testEncodeAndDecodeGenericRecord(Schema encodeSchema, assertTrue(field4 instanceof GenericRecord); GenericRecord field4Record = (GenericRecord) field4; assertEquals(i % 2 == 0, field4Record.getField("field1")); + Object fieldUnableNull = record.getField("fieldUnableNull"); + assertEquals("fieldUnableNull-1-" + i, fieldUnableNull, "fieldUnableNull 1 is " + fieldUnableNull.getClass()); } } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionGenericSchemaProviderTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionGenericSchemaProviderTest.java index 884a674bcd9bd..aedc0bdb6e601 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionGenericSchemaProviderTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionGenericSchemaProviderTest.java @@ -24,10 +24,12 @@ import static org.testng.Assert.assertEquals; import org.apache.pulsar.client.api.schema.GenericSchema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.LookupService; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.SchemaTestUtils; +import org.apache.pulsar.client.tutorial.JsonPojo; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.schema.SchemaInfo; import org.testng.annotations.BeforeMethod; @@ -54,7 +56,7 @@ public void setup() { @Test public void testGetSchema() { CompletableFuture> completableFuture = new CompletableFuture<>(); - SchemaInfo schemaInfo = AvroSchema.of(SchemaTestUtils.Foo.class).getSchemaInfo(); + SchemaInfo schemaInfo = AvroSchema.of(SchemaDefinition.builder().withPojo(SchemaTestUtils.class).build()).getSchemaInfo(); completableFuture.complete(Optional.of(schemaInfo)); when(schemaProvider.getPulsarClient().getLookup() .getSchema( diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleAsyncProducerWithSchema.java b/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleAsyncProducerWithSchema.java index 90758b6edaef4..c838244350a05 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleAsyncProducerWithSchema.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleAsyncProducerWithSchema.java @@ -27,15 +27,15 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.JSONSchema; - @Slf4j public class SampleAsyncProducerWithSchema { public static void main(String[] args) throws IOException { PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("http://localhost:8080").build(); - Producer producer = pulsarClient.newProducer(JSONSchema.of(JsonPojo.class)).topic("persistent://my-property/use/my-ns/my-topic") + Producer producer = pulsarClient.newProducer(JSONSchema.of(SchemaDefinition.builder().withPojo(JsonPojo.class).build())).topic("persistent://my-property/use/my-ns/my-topic") .sendTimeout(3, TimeUnit.SECONDS).create(); List> futures = Lists.newArrayList(); diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleConsumerWithSchema.java b/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleConsumerWithSchema.java index 378033247e78c..1708b84c02717 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleConsumerWithSchema.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/tutorial/SampleConsumerWithSchema.java @@ -23,6 +23,7 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.JSONSchema; public class SampleConsumerWithSchema { @@ -30,7 +31,8 @@ public static void main(String[] args) throws PulsarClientException, JsonProcess PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("http://localhost:8080").build(); - Consumer consumer = pulsarClient.newConsumer(JSONSchema.of(JsonPojo.class)) // + Consumer consumer = pulsarClient.newConsumer(JSONSchema.of + (SchemaDefinition.builder().withPojo(JsonPojo.class).build())) // .topic("persistent://my-property/use/my-ns/my-topic") // .subscriptionName("my-subscription-name").subscribe(); diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java index d8abcc6ee2475..2ac42f46641f1 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java @@ -27,6 +27,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.schema.GenericRecord; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; @@ -124,10 +125,10 @@ private static Schema newSchemaInstance(Class clazz, SchemaType type) return (Schema) Schema.STRING; case AVRO: - return AvroSchema.of(clazz); + return AvroSchema.of(SchemaDefinition.builder().withPojo(clazz).build()); case JSON: - return JSONSchema.of(clazz); + return JSONSchema.of(SchemaDefinition.builder().withPojo(clazz).build()); case KEY_VALUE: return (Schema)Schema.KV_BYTES(); diff --git a/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java b/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java index 0478775bd05dd..984f65828609f 100644 --- a/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java +++ b/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java @@ -31,6 +31,7 @@ import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.schema.GenericRecord; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; import org.apache.pulsar.client.impl.schema.AvroSchema; @@ -108,7 +109,7 @@ public void TestOpenAndWriteSink() throws Exception { obj.setAddress("address_value"); obj.setAge(30); obj.setFlag(true); - AvroSchema schema = AvroSchema.of(Foo.class); + AvroSchema schema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); byte[] bytes = schema.encode(obj); ByteBuf payload = Unpooled.copiedBuffer(bytes); diff --git a/pulsar-io/jdbc/src/test/java/org/apache/pulsar/io/jdbc/JdbcSinkTest.java b/pulsar-io/jdbc/src/test/java/org/apache/pulsar/io/jdbc/JdbcSinkTest.java index 7cbaa57714c18..9ecc91a7af0d8 100644 --- a/pulsar-io/jdbc/src/test/java/org/apache/pulsar/io/jdbc/JdbcSinkTest.java +++ b/pulsar-io/jdbc/src/test/java/org/apache/pulsar/io/jdbc/JdbcSinkTest.java @@ -29,6 +29,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.schema.GenericRecord; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; import org.apache.pulsar.client.impl.schema.AvroSchema; @@ -95,7 +96,7 @@ public void TestOpenAndWriteSink() throws Exception { obj.setField1("ValueOfField1"); obj.setField2("ValueOfField1"); obj.setField3(3); - AvroSchema schema = AvroSchema.of(Foo.class); + AvroSchema schema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); byte[] bytes = schema.encode(obj); ByteBuf payload = Unpooled.copiedBuffer(bytes); diff --git a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java index b6eade22da821..7a62f675a91eb 100644 --- a/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java +++ b/pulsar-sql/presto-pulsar/src/test/java/org/apache/pulsar/sql/presto/TestPulsarConnector.java @@ -41,6 +41,7 @@ import org.apache.pulsar.client.admin.Tenants; import org.apache.pulsar.client.admin.Topics; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.common.api.Commands; @@ -219,21 +220,21 @@ public static class Boo { partitionedTopicsToPartitions.put(PARTITIONED_TOPIC_6.toString(), 7); topicsToSchemas = new HashMap<>(); - topicsToSchemas.put(TOPIC_1.getSchemaName(), AvroSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(TOPIC_2.getSchemaName(), AvroSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(TOPIC_3.getSchemaName(), AvroSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(TOPIC_4.getSchemaName(), JSONSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(TOPIC_5.getSchemaName(), JSONSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(TOPIC_6.getSchemaName(), JSONSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); + topicsToSchemas.put(TOPIC_1.getSchemaName(), AvroSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(TOPIC_2.getSchemaName(), AvroSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(TOPIC_3.getSchemaName(), AvroSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(TOPIC_4.getSchemaName(), JSONSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(TOPIC_5.getSchemaName(), JSONSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(TOPIC_6.getSchemaName(), JSONSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); - topicsToSchemas.put(PARTITIONED_TOPIC_1.getSchemaName(), AvroSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); + topicsToSchemas.put(PARTITIONED_TOPIC_1.getSchemaName(), AvroSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); - topicsToSchemas.put(PARTITIONED_TOPIC_2.getSchemaName(), AvroSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(PARTITIONED_TOPIC_3.getSchemaName(), AvroSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(PARTITIONED_TOPIC_4.getSchemaName(), JSONSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(PARTITIONED_TOPIC_5.getSchemaName(), JSONSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); - topicsToSchemas.put(PARTITIONED_TOPIC_6.getSchemaName(), JSONSchema.of(TestPulsarMetadata.Foo.class).getSchemaInfo()); + topicsToSchemas.put(PARTITIONED_TOPIC_2.getSchemaName(), AvroSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(PARTITIONED_TOPIC_3.getSchemaName(), AvroSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(PARTITIONED_TOPIC_4.getSchemaName(), JSONSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(PARTITIONED_TOPIC_5.getSchemaName(), JSONSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); + topicsToSchemas.put(PARTITIONED_TOPIC_6.getSchemaName(), JSONSchema.of(SchemaDefinition.builder().withPojo(TestPulsarMetadata.Foo.class).build()).getSchemaInfo()); fooTypes = new HashMap<>(); fooTypes.put("field1", IntegerType.INTEGER); @@ -622,7 +623,7 @@ private static List getTopicEntries(String topicSchemaName) { .setProducerName("test-producer").setSequenceId(i) .setPublishTime(currentTimeMs + i).build(); - Schema schema = topicsToSchemas.get(topicSchemaName).getType() == SchemaType.AVRO ? AvroSchema.of(Foo.class) : JSONSchema.of(Foo.class); + Schema schema = topicsToSchemas.get(topicSchemaName).getType() == SchemaType.AVRO ? AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()) : JSONSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); org.apache.pulsar.shade.io.netty.buffer.ByteBuf payload = org.apache.pulsar.shade.io.netty.buffer.Unpooled .copiedBuffer(schema.encode(foo)); diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java index 72d9b0152c01b..da4331abab3f9 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/JdbcSinkTester.java @@ -33,6 +33,7 @@ import lombok.ToString; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.schema.SchemaDefinition; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.tests.integration.topologies.PulsarCluster; import org.testcontainers.containers.MySQLContainer; @@ -60,7 +61,7 @@ public static class Foo { private static final String NAME = "jdbc"; private static final String MYSQL = "mysql"; - private AvroSchema schema = AvroSchema.of(Foo.class); + private AvroSchema schema = AvroSchema.of(SchemaDefinition.builder().withPojo(Foo.class).build()); private String tableName = "test"; private Connection connection;