This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit bcc4b68e7e2b79705bb9b4ae02f7860cd5d3237b
Author: Lari Hotari <[email protected]>
AuthorDate: Thu Nov 27 04:05:10 2025 +0200

    [fix][client] Fix AutoProduceBytesSchema.clone() method (#25015)
    
    (cherry picked from commit 0fd0701d0dc80c8ad2b39b80811360cef197abd8)
---
 .../client/impl/schema/AutoProduceBytesSchema.java |  8 ++-
 .../impl/schema/AutoProduceBytesSchemaTest.java    | 78 ++++++++++++++++++++++
 2 files changed, 85 insertions(+), 1 deletion(-)

diff --git 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.java
 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.java
index cc134b57b21..88809748740 100644
--- 
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.java
+++ 
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchema.java
@@ -49,6 +49,12 @@ public class AutoProduceBytesSchema<T> implements 
Schema<byte[]> {
                                        && schemaInfo.getType() != 
SchemaType.NONE;
     }
 
+    private AutoProduceBytesSchema(AutoProduceBytesSchema<T> other) {
+        this.schema = other.schema != null ? other.schema.clone() : null;
+        this.userProvidedSchema = other.userProvidedSchema;
+        this.requireSchemaValidation = other.requireSchemaValidation;
+    }
+
     public void setSchema(Schema<T> schema) {
         this.schema = schema;
         this.requireSchemaValidation = schema.getSchemaInfo() != null
@@ -121,6 +127,6 @@ public class AutoProduceBytesSchema<T> implements 
Schema<byte[]> {
 
     @Override
     public Schema<byte[]> clone() {
-        return new AutoProduceBytesSchema<>(schema.clone());
+        return new AutoProduceBytesSchema<>(this);
     }
 }
diff --git 
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchemaTest.java
 
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchemaTest.java
new file mode 100644
index 00000000000..b67763d5d85
--- /dev/null
+++ 
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchemaTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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 org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertTrue;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.annotations.Test;
+
+public class AutoProduceBytesSchemaTest {
+    @Test
+    public void testClone() {
+        // test user provided schema
+        Schema<String> stringSchema = Schema.STRING;
+        AutoProduceBytesSchema<String> schema1 = new 
AutoProduceBytesSchema<>(stringSchema);
+        Schema<byte[]> clone1 = schema1.clone();
+
+        assertNotSame(schema1, clone1);
+        assertTrue(clone1 instanceof AutoProduceBytesSchema);
+        AutoProduceBytesSchema<String> castedClone1 = 
(AutoProduceBytesSchema<String>) clone1;
+        assertTrue(castedClone1.hasUserProvidedSchema());
+        assertTrue(castedClone1.schemaInitialized());
+        assertEquals(castedClone1.getSchemaInfo().getType(), 
SchemaType.STRING);
+
+        // test no user provided schema
+        AutoProduceBytesSchema<String> schema2 = new 
AutoProduceBytesSchema<>();
+        Schema<byte[]> clone2 = schema2.clone();
+
+        assertNotSame(schema2, clone2);
+        assertTrue(clone2 instanceof AutoProduceBytesSchema);
+        AutoProduceBytesSchema<String> castedClone2 = 
(AutoProduceBytesSchema<String>) clone2;
+        assertFalse(castedClone2.hasUserProvidedSchema());
+        assertFalse(castedClone2.schemaInitialized());
+
+        // test no user provided schema after setSchema
+        AutoProduceBytesSchema<String> schema3 = new 
AutoProduceBytesSchema<>();
+        schema3.setSchema(stringSchema);
+        Schema<byte[]> clone3 = schema3.clone();
+
+        assertNotSame(schema3, clone3);
+        assertTrue(clone3 instanceof AutoProduceBytesSchema);
+        AutoProduceBytesSchema<String> castedClone3 = 
(AutoProduceBytesSchema<String>) clone3;
+        assertFalse(castedClone3.hasUserProvidedSchema());
+        assertTrue(castedClone3.schemaInitialized());
+        assertEquals(castedClone3.getSchemaInfo().getType(), 
SchemaType.STRING);
+    }
+
+    @Test
+    public void testInnerSchemaGetsCloned() {
+        Schema<String> stringSchema = spy(Schema.STRING);
+        AutoProduceBytesSchema<String> schema = new 
AutoProduceBytesSchema<>(stringSchema);
+        schema.clone();
+        verify(stringSchema, times(1)).clone();
+    }
+
+}
\ No newline at end of file

Reply via email to