[Issue 11007] Schema.NATIVE_AVRO: add a version of AUTO_PRODUCE_BYTES that doesn't validate the message in encode

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java

@@ -426,6 +426,19 @@ static Schema<byte[]> AUTO_PRODUCE_BYTES(Schema<?> schema) {
         return DefaultImplementation.newAutoProduceSchema(schema);
     }
 
+    /**
+     * Create a schema instance that accepts a serialized Avro payload
+     * without validating it against the schema specified. 
+     * It can be useful when migrating data from existing event or message stores.
+     *
+     * @return the auto schema instance
+     * @since 2.9.0
+     * @see #AUTO_PRODUCE_BYTES()
+     */
+    static Schema<byte[]> AUTO_PRODUCE_VALIDATED_AVRO_BYTES(Schema<?> schema) {
+        return DefaultImplementation.newAutoProduceValidatedAvroSchema(schema);
+    }
+
     // CHECKSTYLE.ON: MethodName
 
     static Schema<?> getSchema(SchemaInfo schemaInfo) {

pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java

@@ -298,6 +298,13 @@ public static Schema<byte[]> newAutoProduceSchema(Schema<?> schema) {
                         .newInstance(schema));
     }
 
+    public static Schema<byte[]> newAutoProduceValidatedAvroSchema(Schema<?> schema) {
+        return catchExceptions(
+                () -> (Schema<byte[]>) getConstructor(
+                    "org.apache.pulsar.client.impl.schema.AutoProduceValidatedAvroBytesSchema", Schema.class)
+                        .newInstance(schema));
+    }
+
     public static Schema<KeyValue<byte[], byte[]>> newKeyValueBytesSchema() {
         return catchExceptions(
                 () -> (Schema<KeyValue<byte[], byte[]>>) getStaticMethod(

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceValidatedAvroBytesSchema.java

@@ -0,0 +1,74 @@
+/**
+ * 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.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.internal.DefaultImplementation;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import java.util.Optional;
+
+/**
+ * Auto detect schema.
+ */
+public class AutoProduceValidatedAvroBytesSchema<T> extends AutoProduceBytesSchema<T> {
+
+    private Schema<T> schema;
+
+
+    public AutoProduceValidatedAvroBytesSchema(org.apache.avro.Schema schema) {
+        SchemaDefinition schemaDefinition = SchemaDefinition.builder().withJsonDef(schema.toString(false)).build();
+
+        setSchema(DefaultImplementation.newAvroSchema(schemaDefinition));
+    }
+
+    public AutoProduceValidatedAvroBytesSchema(Schema<T> schema) {
+        setSchema(schema);
+    }
+
+    public void setSchema(Schema<T> schema) {
+        if (SchemaType.AVRO != schema.getSchemaInfo().getType()) {
+            throw new IllegalArgumentException("Provided schema is not an Avro type.");
+        }
+        this.schema = schema;
+    }
+
+    private void ensureSchemaInitialized() {
+        checkState(schemaInitialized(), "Schema is not initialized before used");
+    }
+
+    @Override
+    public byte[] encode(byte[] message) {
+        ensureSchemaInitialized();
+
+        return message;
+    }
+
+    @Override
+    public byte[] decode(byte[] bytes, byte[] schemaVersion) {
+        ensureSchemaInitialized();
+
+        return bytes;
+    }
+
+}