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 24573d5bc9a45..e2c8747564969 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 @@ -18,6 +18,8 @@ */ package org.apache.pulsar.client.api; +import static org.testng.Assert.fail; + import com.google.common.base.MoreObjects; import com.google.common.collect.Sets; import java.time.Clock; @@ -459,7 +461,7 @@ public void testAvroProducerAndAutoSchemaConsumer() throws Exception { } Consumer consumer = pulsarClient - .newConsumer(Schema.AUTO()) + .newConsumer(Schema.AUTO_CONSUME()) .topic("persistent://my-property/use/my-ns/my-topic1") .subscriptionName("my-subscriber-name") .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) @@ -507,7 +509,7 @@ public void testAvroProducerAndAutoSchemaReader() throws Exception { } Reader reader = pulsarClient - .newReader(Schema.AUTO()) + .newReader(Schema.AUTO_CONSUME()) .topic("persistent://my-property/use/my-ns/my-topic1") .startMessageId(MessageId.earliest) .create(); @@ -535,4 +537,76 @@ public void testAvroProducerAndAutoSchemaReader() throws Exception { } + @Test + public void testAutoBytesProducer() throws Exception { + log.info("-- Starting {} test --", methodName); + + AvroSchema avroSchema = + AvroSchema.of(AvroEncodedPojo.class); + + try (Producer producer = pulsarClient + .newProducer(avroSchema) + .topic("persistent://my-property/use/my-ns/my-topic1") + .create()) { + for (int i = 0; i < 10; i++) { + String message = "my-message-" + i; + producer.send(new AvroEncodedPojo(message)); + } + } + + try (Producer producer = pulsarClient + .newProducer(Schema.AUTO_PRODUCE_BYTES()) + .topic("persistent://my-property/use/my-ns/my-topic1") + .create()) { + // try to produce junk data + for (int i = 10; i < 20; i++) { + String message = "my-message-" + i; + byte[] data = avroSchema.encode(new AvroEncodedPojo(message)); + byte[] junkData = new byte[data.length / 2]; + System.arraycopy(data, 0, junkData, 0, junkData.length); + try { + producer.send(junkData); + fail("Should fail on sending junk data"); + } catch (SchemaSerializationException sse) { + // expected + } + } + + for (int i = 10; i < 20; i++) { + String message = "my-message-" + i; + byte[] data = avroSchema.encode(new AvroEncodedPojo(message)); + producer.send(data); + } + } + + Consumer consumer = pulsarClient + .newConsumer(Schema.AUTO_CONSUME()) + .topic("persistent://my-property/use/my-ns/my-topic1") + .subscriptionName("my-subscriber-name") + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + + Message msg = null; + Set messageSet = Sets.newHashSet(); + for (int i = 0; i < 20; i++) { + msg = consumer.receive(5, TimeUnit.SECONDS); + GenericRecord receivedMessage = msg.getValue(); + log.debug("Received message: [{}]", receivedMessage); + String expectedMessage = "my-message-" + i; + String actualMessage = (String) receivedMessage.getField("message"); + testMessageOrderAndDuplicates(messageSet, actualMessage, expectedMessage); + } + // Acknowledge the consumption of all messages at once + consumer.acknowledgeCumulative(msg); + consumer.close(); + + SchemaRegistry.SchemaAndMetadata storedSchema = pulsar.getSchemaRegistryService() + .getSchema("my-property/my-ns/my-topic1") + .get(); + + Assert.assertEquals(storedSchema.schema.getData(), avroSchema.getSchemaInfo().getSchema()); + + log.info("-- Exiting {} test --", methodName); + + } } diff --git a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java index 113f26c50ef00..5432d43108596 100644 --- a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java +++ b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/api/Schema.java @@ -19,12 +19,20 @@ package org.apache.pulsar.client.api; import org.apache.pulsar.client.api.schema.GenericRecord; -import org.apache.pulsar.client.impl.schema.AutoSchema; +import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; +import org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema; import org.apache.pulsar.client.impl.schema.AvroSchema; +import org.apache.pulsar.client.impl.schema.ByteSchema; import org.apache.pulsar.client.impl.schema.BytesSchema; +import org.apache.pulsar.client.impl.schema.DoubleSchema; +import org.apache.pulsar.client.impl.schema.FloatSchema; +import org.apache.pulsar.client.impl.schema.IntSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.client.impl.schema.LongSchema; import org.apache.pulsar.client.impl.schema.ProtobufSchema; +import org.apache.pulsar.client.impl.schema.ShortSchema; import org.apache.pulsar.client.impl.schema.StringSchema; +import org.apache.pulsar.client.impl.schema.generic.GenericSchema; import org.apache.pulsar.common.schema.SchemaInfo; /** @@ -80,7 +88,43 @@ static Schema JSON(Class clazz) { return JSONSchema.of(clazz); } + @Deprecated static Schema AUTO() { - return new AutoSchema(); + return AUTO_CONSUME(); + } + + static Schema AUTO_CONSUME() { + return new AutoConsumeSchema(); + } + + static Schema AUTO_PRODUCE_BYTES() { + return new AutoProduceBytesSchema(); + } + + static Schema getSchema(SchemaInfo schemaInfo) { + switch (schemaInfo.getType()) { + case INT8: + return ByteSchema.of(); + case INT16: + return ShortSchema.of(); + case INT32: + return IntSchema.of(); + case INT64: + return LongSchema.of(); + case STRING: + return StringSchema.utf8(); + case FLOAT: + return FloatSchema.of(); + case DOUBLE: + return DoubleSchema.of(); + case BYTES: + return BytesSchema.of(); + case JSON: + case AVRO: + return GenericSchema.of(schemaInfo); + default: + throw new IllegalArgumentException("Retrieve schema instance from schema info for type '" + + schemaInfo.getType() + "' is not supported yet"); + } } } diff --git a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoSchema.java b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java similarity index 96% rename from pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoSchema.java rename to pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java index 5bf92b71cbb05..dc90931d4f4d9 100644 --- a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoSchema.java +++ b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java @@ -28,7 +28,7 @@ /** * Auto detect schema. */ -public class AutoSchema implements Schema { +public class AutoConsumeSchema implements Schema { private Schema schema; diff --git a/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.java b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.java new file mode 100644 index 0000000000000..36520a9ee78d2 --- /dev/null +++ b/pulsar-client-schema/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.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.client.impl.schema; + + +import static com.google.common.base.Preconditions.checkState; + +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.schema.SchemaInfo; + +/** + * Auto detect schema. + */ +public class AutoProduceBytesSchema implements Schema { + + private Schema schema; + + public void setSchema(Schema schema) { + this.schema = schema; + } + + private void ensureSchemaInitialized() { + checkState(null != schema, "Schema is not initialized before used"); + } + + @Override + public byte[] encode(byte[] message) { + ensureSchemaInitialized(); + + // verify if the message can be decoded by the underlying schema + schema.decode(message); + + return message; + } + + @Override + public byte[] decode(byte[] bytes) { + ensureSchemaInitialized(); + + // verify the message can be detected by the underlying schema + schema.decode(bytes); + + return bytes; + } + + @Override + public SchemaInfo getSchemaInfo() { + ensureSchemaInitialized(); + + return schema.getSchemaInfo(); + } +} diff --git a/pulsar-client-schema/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java b/pulsar-client-schema/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java index f07c92887007e..f00d05da0bddb 100644 --- a/pulsar-client-schema/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java +++ b/pulsar-client-schema/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java @@ -25,7 +25,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.schema.GenericRecord; -import org.apache.pulsar.client.impl.schema.AutoSchema; +import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; import org.apache.pulsar.client.schema.SchemaTestUtils.Bar; import org.apache.pulsar.client.schema.SchemaTestUtils.Foo; import org.testng.annotations.Test; @@ -53,7 +53,7 @@ public void testGenericJsonSchema() { @Test public void testAutoAvroSchema() { Schema encodeSchema = Schema.AVRO(Foo.class); - AutoSchema decodeSchema = new AutoSchema(); + AutoConsumeSchema decodeSchema = new AutoConsumeSchema(); decodeSchema.setSchema(GenericSchema.of(encodeSchema.getSchemaInfo())); testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema); } @@ -61,7 +61,7 @@ public void testAutoAvroSchema() { @Test public void testAutoJsonSchema() { Schema encodeSchema = Schema.JSON(Foo.class); - AutoSchema decodeSchema = new AutoSchema(); + AutoConsumeSchema decodeSchema = new AutoConsumeSchema(); decodeSchema.setSchema(GenericSchema.of(encodeSchema.getSchemaInfo())); testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index 9640c09dd7a5a..bac6a85164026 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -62,7 +62,8 @@ import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; -import org.apache.pulsar.client.impl.schema.AutoSchema; +import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; +import org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema; import org.apache.pulsar.client.impl.schema.generic.GenericSchema; import org.apache.pulsar.client.util.ExecutorProvider; import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace.Mode; @@ -250,12 +251,12 @@ public CompletableFuture> createProducerAsync(ProducerConfigurat ProducerInterceptors interceptors) { if (conf == null) { return FutureUtil.failedFuture( - new PulsarClientException.InvalidConfigurationException("Producer configuration undefined")); + new PulsarClientException.InvalidConfigurationException("Producer configuration undefined")); } - if (schema instanceof AutoSchema) { + if (schema instanceof AutoConsumeSchema) { return FutureUtil.failedFuture( - new PulsarClientException.InvalidConfigurationException("AutoSchema is only used by consumers to detect schemas automatically")); + new PulsarClientException.InvalidConfigurationException("AutoConsumeSchema is only used by consumers to detect schemas automatically")); } if (state.get() != State.Open) { @@ -266,9 +267,30 @@ public CompletableFuture> createProducerAsync(ProducerConfigurat if (!TopicName.isValid(topic)) { return FutureUtil.failedFuture( - new PulsarClientException.InvalidTopicNameException("Invalid topic name: '" + topic + "'")); + new PulsarClientException.InvalidTopicNameException("Invalid topic name: '" + topic + "'")); } + if (schema instanceof AutoProduceBytesSchema) { + AutoProduceBytesSchema autoProduceBytesSchema = (AutoProduceBytesSchema) schema; + return lookup.getSchema(TopicName.get(conf.getTopicName())) + .thenCompose(schemaInfoOptional -> { + if (schemaInfoOptional.isPresent()) { + autoProduceBytesSchema.setSchema(Schema.getSchema(schemaInfoOptional.get())); + } else { + autoProduceBytesSchema.setSchema(Schema.BYTES); + } + return createProducerAsync(topic, conf, schema, interceptors); + }); + } else { + return createProducerAsync(topic, conf, schema, interceptors); + } + + } + + private CompletableFuture> createProducerAsync(String topic, + ProducerConfigurationData conf, + Schema schema, + ProducerInterceptors interceptors) { CompletableFuture> producerCreatedFuture = new CompletableFuture<>(); getPartitionedTopicMetadata(topic).thenAccept(metadata -> { @@ -392,15 +414,15 @@ public CompletableFuture> subscribeAsync(ConsumerConfigurationDa } private CompletableFuture> singleTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { - if (schema instanceof AutoSchema) { - AutoSchema autoSchema = (AutoSchema) schema; + if (schema instanceof AutoConsumeSchema) { + AutoConsumeSchema autoConsumeSchema = (AutoConsumeSchema) schema; return lookup.getSchema(TopicName.get(conf.getSingleTopic())) .thenCompose(schemaInfoOptional -> { if (schemaInfoOptional.isPresent() && schemaInfoOptional.get().getType() == SchemaType.AVRO) { GenericSchema genericSchema = GenericSchema.of(schemaInfoOptional.get()); log.info("Auto detected schema for topic {} : {}", conf.getSingleTopic(), new String(schemaInfoOptional.get().getSchema(), UTF_8)); - autoSchema.setSchema(genericSchema); + autoConsumeSchema.setSchema(genericSchema); return doSingleTopicSubscribeAsync(conf, schema, interceptors); } else { return FutureUtil.failedFuture( @@ -546,15 +568,15 @@ public CompletableFuture> createReaderAsync(ReaderConfigurationDa } public CompletableFuture> createReaderAsync(ReaderConfigurationData conf, Schema schema) { - if (schema instanceof AutoSchema) { - AutoSchema autoSchema = (AutoSchema) schema; + if (schema instanceof AutoConsumeSchema) { + AutoConsumeSchema autoConsumeSchema = (AutoConsumeSchema) schema; return lookup.getSchema(TopicName.get(conf.getTopicName())) .thenCompose(schemaInfoOptional -> { if (schemaInfoOptional.isPresent() && schemaInfoOptional.get().getType() == SchemaType.AVRO) { GenericSchema genericSchema = GenericSchema.of(schemaInfoOptional.get()); log.info("Auto detected schema for topic {} : {}", conf.getTopicName(), new String(schemaInfoOptional.get().getSchema(), UTF_8)); - autoSchema.setSchema(genericSchema); + autoConsumeSchema.setSchema(genericSchema); return doCreateReaderAsync(conf, schema); } else { return FutureUtil.failedFuture( diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java b/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java index a19d587e17013..4b4e576b6d3c6 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java @@ -85,5 +85,16 @@ public enum SchemaType { /** * Auto Detect Schema Type. */ - AUTO + @Deprecated + AUTO, + + /** + * Auto Consume Type. + */ + AUTO_CONSUME, + + /** + * Auto Publish Type. + */ + AUTO_PUBLISH } 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 db9e880b68b94..cbf97f1330cbc 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 @@ -77,7 +77,7 @@ public Schema getSchema(String topic, Class clazz, SchemaType schemaType) */ private SchemaType getSchemaTypeOrDefault(String topic, Class clazz) { if (GenericRecord.class.isAssignableFrom(clazz)) { - return SchemaType.AUTO; + return SchemaType.AUTO_CONSUME; } else if (byte[].class.equals(clazz)) { // if function uses bytes, we should ignore return SchemaType.NONE; @@ -96,7 +96,7 @@ private static SchemaType getDefaultSchemaType(Class clazz) { return SchemaType.NONE; } else if (GenericRecord.class.isAssignableFrom(clazz)) { // the function is taking generic record, so we do auto schema detection - return SchemaType.AUTO; + return SchemaType.AUTO_CONSUME; } else if (String.class.equals(clazz)) { // If type is String, then we use schema type string, otherwise we fallback on default schema return SchemaType.STRING; @@ -113,8 +113,9 @@ private static Schema newSchemaInstance(Class clazz, SchemaType type) case NONE: return (Schema) Schema.BYTES; + case AUTO_CONSUME: case AUTO: - return (Schema) Schema.AUTO(); + return (Schema) Schema.AUTO_CONSUME(); case STRING: return (Schema) Schema.STRING; diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java index b92ae2747465c..45a861bce1012 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java @@ -71,6 +71,7 @@ import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.Reader; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.io.ConnectorDefinition; import org.apache.pulsar.common.policies.data.ErrorData; import org.apache.pulsar.common.policies.data.TenantInfo; @@ -635,13 +636,15 @@ public Response triggerFunction(final String tenant, final String namespace, fin return Response.status(Status.BAD_REQUEST).build(); } String outputTopic = functionMetaData.getFunctionDetails().getSink().getTopic(); - Reader reader = null; - Producer producer = null; + Reader reader = null; + Producer producer = null; try { if (outputTopic != null && !outputTopic.isEmpty()) { reader = worker().getClient().newReader().topic(outputTopic).startMessageId(MessageId.latest).create(); } - producer = worker().getClient().newProducer().topic(inputTopicToWrite).create(); + producer = worker().getClient().newProducer(Schema.AUTO_PRODUCE_BYTES()) + .topic(inputTopicToWrite) + .create(); byte[] targetArray; if (uploadedInputStream != null) { targetArray = new byte[uploadedInputStream.available()]; 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 5067f55de7780..24a7c6e88359a 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 @@ -30,7 +30,7 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.impl.MessageImpl; -import org.apache.pulsar.client.impl.schema.AutoSchema; +import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.generic.GenericSchema; import org.apache.pulsar.functions.api.Record; @@ -99,10 +99,10 @@ public void TestOpenAndWriteSink() throws Exception { byte[] bytes = schema.encode(obj); ByteBuf payload = Unpooled.copiedBuffer(bytes); - AutoSchema autoSchema = new AutoSchema(); - autoSchema.setSchema(GenericSchema.of(schema.getSchemaInfo())); + AutoConsumeSchema autoConsumeSchema = new AutoConsumeSchema(); + autoConsumeSchema.setSchema(GenericSchema.of(schema.getSchemaInfo())); - Message message = new MessageImpl("fake_topic_name", "77:777", conf, payload, autoSchema); + Message message = new MessageImpl("fake_topic_name", "77:777", conf, payload, autoConsumeSchema); Record record = PulsarRecord.builder() .message(message) .topicName("fake_topic_name")