You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/11/27 09:12:58 UTC
(pulsar) branch branch-4.0 updated (31f1ff801fc -> c5efa8f29e3)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 31f1ff801fc [improve][client] Test no exception could be thrown for invalid epoch in message (#25013)
new c7abb70de42 [fix][broker] Add schema version in rest produce api (#25004)
new 991ce21048c [improve] Eliminate unnecessary duplicate schema lookups for partitioned topics in client and geo-replication (#25011)
new 7a8f92fa98e [improve][broker]Remove the warn log that frequently prints (#25018)
new bcc4b68e7e2 [fix][client] Fix AutoProduceBytesSchema.clone() method (#25015)
new 0c513139ebb [fix][client] Fix thread-safety of AutoProduceBytesSchema (#25014)
new c5efa8f29e3 [fix] Upgrade gson to 2.13.2 (#25022)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
distribution/server/src/assemble/LICENSE.bin.txt | 2 +-
distribution/shell/src/assemble/LICENSE.bin.txt | 2 +-
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 -
pom.xml | 2 +-
.../org/apache/pulsar/broker/rest/TopicsBase.java | 7 +-
.../service/persistent/PersistentReplicator.java | 5 +-
.../org/apache/pulsar/broker/admin/TopicsTest.java | 63 +++++++++++++++++
.../pulsar/client/impl/PulsarClientImpl.java | 5 +-
.../client/impl/schema/AutoProduceBytesSchema.java | 24 ++++---
.../impl/schema/AutoProduceBytesSchemaTest.java | 78 ++++++++++++++++++++++
10 files changed, 171 insertions(+), 19 deletions(-)
create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoProduceBytesSchemaTest.java
(pulsar) 02/06: [improve] Eliminate unnecessary duplicate schema lookups for partitioned topics in client and geo-replication (#25011)
Posted by lh...@apache.org.
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 991ce21048c5f1167033f07ca33682f681a41b79
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Nov 26 07:38:57 2025 +0200
[improve] Eliminate unnecessary duplicate schema lookups for partitioned topics in client and geo-replication (#25011)
(cherry picked from commit 163f35fd77af5d2ac6191f837a9d726c172fb0bd)
---
.../pulsar/broker/service/persistent/PersistentReplicator.java | 5 ++++-
.../main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java | 5 +++--
2 files changed, 7 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index bda5acce236..e0a31476fc9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -67,6 +67,7 @@ import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.SendCallback;
import org.apache.pulsar.common.api.proto.MarkerType;
+import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.stats.Rate;
@@ -80,6 +81,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
protected final PersistentTopic topic;
protected final ManagedCursor cursor;
+ protected final String localSchemaTopicName;
protected Optional<DispatchRateLimiter> dispatchRateLimiter = Optional.empty();
private final Object dispatchRateLimiterLock = new Object();
@@ -123,6 +125,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
super(localCluster, localTopic, remoteCluster, remoteTopic, localTopic.getReplicatorPrefix(),
brokerService, replicationClient);
this.topic = localTopic;
+ this.localSchemaTopicName = TopicName.getPartitionedTopicName(localTopicName).toString();
this.cursor = Objects.requireNonNull(cursor);
this.expiryMonitor = new PersistentMessageExpiryMonitor(localTopic,
Codec.decode(cursor.getName()), cursor, null);
@@ -378,7 +381,7 @@ public abstract class PersistentReplicator extends AbstractReplicator
if (msg.getSchemaVersion() == null || msg.getSchemaVersion().length == 0) {
return CompletableFuture.completedFuture(null);
}
- return client.getSchemaProviderLoadingCache().get(localTopicName)
+ return client.getSchemaProviderLoadingCache().get(localSchemaTopicName)
.getSchemaByVersion(msg.getSchemaVersion());
}
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 f3edafc8efd..cac6617c9dd 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
@@ -1335,10 +1335,11 @@ public class PulsarClientImpl implements PulsarClient {
String topicName) {
if (schema != null && schema.supportSchemaVersioning()) {
final SchemaInfoProvider schemaInfoProvider;
+ String schemaTopicName = TopicName.getPartitionedTopicName(topicName).toString();
try {
- schemaInfoProvider = pulsarClientImpl.getSchemaProviderLoadingCache().get(topicName);
+ schemaInfoProvider = pulsarClientImpl.getSchemaProviderLoadingCache().get(schemaTopicName);
} catch (ExecutionException e) {
- log.error("Failed to load schema info provider for topic {}", topicName, e);
+ log.error("Failed to load schema info provider for topic {}", schemaTopicName, e);
return FutureUtil.failedFuture(e.getCause());
}
schema = schema.clone();
(pulsar) 05/06: [fix][client] Fix thread-safety of AutoProduceBytesSchema (#25014)
Posted by lh...@apache.org.
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 0c513139ebbf9c41f885090754cb29f749c2504d
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Nov 27 04:08:41 2025 +0200
[fix][client] Fix thread-safety of AutoProduceBytesSchema (#25014)
(cherry picked from commit d3d5963bdcfaa7b03bef4d14bebba4924e2b5cf7)
---
.../client/impl/schema/AutoProduceBytesSchema.java | 16 +++++++++-------
1 file changed, 9 insertions(+), 7 deletions(-)
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 88809748740..927ae94e027 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
@@ -33,11 +33,12 @@ import org.apache.pulsar.common.schema.SchemaType;
public class AutoProduceBytesSchema<T> implements Schema<byte[]> {
@Setter
- private boolean requireSchemaValidation = true;
- private Schema<T> schema;
- private boolean userProvidedSchema;
+ private volatile boolean requireSchemaValidation = true;
+ private volatile Schema<T> schema;
+ private final boolean userProvidedSchema;
public AutoProduceBytesSchema() {
+ this.userProvidedSchema = false;
}
public AutoProduceBytesSchema(Schema<T> schema) {
@@ -87,11 +88,12 @@ public class AutoProduceBytesSchema<T> implements Schema<byte[]> {
if (requireSchemaValidation) {
// verify if the message can be decoded by the underlying schema
- if (schema instanceof KeyValueSchema
- && ((KeyValueSchema) schema).getKeyValueEncodingType().equals(KeyValueEncodingType.SEPARATED)) {
- ((KeyValueSchema) schema).getValueSchema().validate(message);
+ Schema<T> localSchema = schema;
+ if (localSchema instanceof KeyValueSchema && ((KeyValueSchema) localSchema).getKeyValueEncodingType()
+ .equals(KeyValueEncodingType.SEPARATED)) {
+ ((KeyValueSchema) localSchema).getValueSchema().validate(message);
} else {
- schema.validate(message);
+ localSchema.validate(message);
}
}
(pulsar) 03/06: [improve][broker]Remove the warn log that frequently prints (#25018)
Posted by lh...@apache.org.
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 7a8f92fa98e8a83e8e3567fc6ddda23c74db798a
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Wed Nov 26 15:29:06 2025 +0800
[improve][broker]Remove the warn log that frequently prints (#25018)
(cherry picked from commit a0637a6c15e2946f459eb5828a79a27d601012b2)
---
.../main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 2 --
1 file changed, 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 6a7844713d5..eb023f4c2df 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -3734,8 +3734,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
|| pos2.getLedgerId() < getFirstPosition().getLedgerId()
|| pos1.getLedgerId() > getLastPosition().getLedgerId()
|| pos2.getLedgerId() > getLastPosition().getLedgerId()) {
- log.warn("[{}] Comparing un-exist position {} and {}", name, pos1, pos2,
- new IllegalArgumentException("Comparing un-exist position"));
return pos1.compareTo(pos2);
}
if (pos1.getLedgerId() == pos2.getLedgerId()) {
(pulsar) 01/06: [fix][broker] Add schema version in rest produce api (#25004)
Posted by lh...@apache.org.
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 c7abb70de42ce9739ef23facb105232522016603
Author: Oneby <44...@users.noreply.github.com>
AuthorDate: Wed Nov 26 13:24:04 2025 +0800
[fix][broker] Add schema version in rest produce api (#25004)
(cherry picked from commit a44e60da13b17f9ec815b4ad413e1b7ff8010e6d)
---
.../org/apache/pulsar/broker/rest/TopicsBase.java | 7 +--
.../org/apache/pulsar/broker/admin/TopicsTest.java | 63 ++++++++++++++++++++++
2 files changed, 67 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
index 47067c55005..425e715a1e5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/rest/TopicsBase.java
@@ -196,7 +196,7 @@ public class TopicsBase extends PersistentTopicsBase {
try {
String producerName = (null == request.getProducerName() || request.getProducerName().isEmpty())
? defaultProducerName : request.getProducerName();
- List<Message> messages = buildMessage(request, schema, producerName, topicName);
+ List<Message> messages = buildMessage(request, schema, producerName, topicName, schemaVersion);
List<CompletableFuture<Position>> publishResults = new ArrayList<>();
List<ProducerAck> produceMessageResults = new ArrayList<>();
for (int index = 0; index < messages.size(); index++) {
@@ -237,7 +237,7 @@ public class TopicsBase extends PersistentTopicsBase {
try {
String producerName = (null == request.getProducerName() || request.getProducerName().isEmpty())
? defaultProducerName : request.getProducerName();
- List<Message> messages = buildMessage(request, schema, producerName, topicName);
+ List<Message> messages = buildMessage(request, schema, producerName, topicName, schemaVersion);
List<CompletableFuture<Position>> publishResults = new ArrayList<>();
List<ProducerAck> produceMessageResults = new ArrayList<>();
// Try to publish messages to all partitions this broker owns in round robin mode.
@@ -627,7 +627,7 @@ public class TopicsBase extends PersistentTopicsBase {
// Build pulsar message from REST request.
private List<Message> buildMessage(ProducerMessages producerMessages, Schema schema,
- String producerName, TopicName topicName) {
+ String producerName, TopicName topicName, SchemaVersion schemaVersion) {
List<ProducerMessage> messages;
List<Message> pulsarMessages = new ArrayList<>();
@@ -637,6 +637,7 @@ public class TopicsBase extends PersistentTopicsBase {
messageMetadata.setProducerName(producerName);
messageMetadata.setPublishTime(System.currentTimeMillis());
messageMetadata.setSequenceId(message.getSequenceId());
+ messageMetadata.setSchemaVersion(schemaVersion.bytes());
if (null != message.getReplicationClusters()) {
messageMetadata.addAllReplicateTos(message.getReplicationClusters());
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java
index 72426110224..bcbd133ce5c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicsTest.java
@@ -86,6 +86,7 @@ import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.websocket.data.ProducerAcks;
import org.apache.pulsar.websocket.data.ProducerMessage;
@@ -868,4 +869,66 @@ public class TopicsTest extends MockedPulsarServiceBaseTest {
+ "kv.encoding.type=SEPARATED, key.schema.type=STRING}) to topic persistent:"
+ "//my-tenant/my-namespace/my-topic"));
}
+
+ @Test
+ public void testProduceWithAutoConsumeSchema() throws Exception {
+ String topicName = "persistent://" + testTenant + "/" + testNamespace + "/" + testTopicName;
+ admin.topics().createNonPartitionedTopic(topicName);
+ GenericSchema jsonSchema = GenericJsonSchema.of(JSONSchema.of(SchemaDefinition.builder()
+ .withPojo(PC.class).build()).getSchemaInfo());
+ // use producer to create topic schema
+ Producer producer = pulsarClient.newProducer(jsonSchema).topic(topicName).create();
+ producer.close();
+
+ PC pc = new PC("dell", "alienware", 2021, GPU.AMD,
+ new Seller("WA", "main street", 98004));
+ PC anotherPc = new PC("asus", "rog", 2020, GPU.NVIDIA,
+ new Seller("CA", "back street", 90232));
+
+ @Cleanup
+ Consumer<?> consumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME())
+ .topic(topicName)
+ .subscriptionName("auto-schema-sub")
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ SchemaInfoWithVersion schemaInfoWithVersion = admin.schemas().getSchemaInfoWithVersion(topicName);
+ AsyncResponse asyncResponse = mock(AsyncResponse.class);
+ ProducerMessages producerMessages = new ProducerMessages();
+ producerMessages.setSchemaVersion(schemaInfoWithVersion.getVersion());
+ String message = "["
+ + "{\"key\":\"my-key\",\"payload\":\""
+ + ObjectMapperFactory.getMapper().writer().writeValueAsString(pc).replace("\"", "\\\"")
+ + "\",\"eventTime\":1603045262772,\"sequenceId\":1},"
+ + "{\"key\":\"my-key\",\"payload\":\""
+ + ObjectMapperFactory.getMapper().writer().writeValueAsString(anotherPc).replace("\"", "\\\"")
+ + "\",\"eventTime\":1603045262772,\"sequenceId\":2}]";
+ producerMessages.setMessages(createMessages(message));
+ topics.produceOnPersistentTopic(asyncResponse, testTenant, testNamespace, testTopicName, false,
+ producerMessages);
+ ArgumentCaptor<Response> responseCaptor = ArgumentCaptor.forClass(Response.class);
+ verify(asyncResponse, timeout(5000).times(1)).resume(responseCaptor.capture());
+ Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.OK.getStatusCode());
+
+ List<PC> expected = Arrays.asList(pc, anotherPc);
+ for (int i = 0; i < 2; i++) {
+ Message<?> msg = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertTrue(msg.getValue() instanceof GenericJsonRecord);
+ GenericJsonRecord genericJsonRecord = (GenericJsonRecord) msg.getValue();
+ Assert.assertEquals(genericJsonRecord.getField("brand"), expected.get(i).getBrand());
+ Assert.assertEquals(genericJsonRecord.getField("model"), expected.get(i).getModel());
+ Number year = (Number) genericJsonRecord.getField("year");
+ Assert.assertEquals(year.intValue(), expected.get(i).getYear());
+ Assert.assertEquals(genericJsonRecord.getField("gpu"), expected.get(i).getGpu().name());
+ Object seller = genericJsonRecord.getField("seller");
+ Assert.assertTrue(seller instanceof GenericJsonRecord);
+ GenericJsonRecord sellerGenericJsonRecord = (GenericJsonRecord) seller;
+ Assert.assertEquals(sellerGenericJsonRecord.getField("state"), expected.get(i).getSeller().getState());
+ Assert.assertEquals(sellerGenericJsonRecord.getField("street"), expected.get(i).getSeller().getStreet());
+ Number zipCode = (Number) sellerGenericJsonRecord.getField("zipCode");
+ Assert.assertEquals(zipCode.longValue(), expected.get(i).getSeller().getZipCode());
+ }
+ }
+
}
(pulsar) 04/06: [fix][client] Fix AutoProduceBytesSchema.clone() method (#25015)
Posted by lh...@apache.org.
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 <lh...@users.noreply.github.com>
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