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 2026/08/27 08:41:58 UTC
(pulsar) branch branch-4.2 updated: [fix][test][branch-4.2] Fix connector tests broken by the Avro 1.12.2 upgrade
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 86c8345d1e37 [fix][test][branch-4.2] Fix connector tests broken by the Avro 1.12.2 upgrade
86c8345d1e37 is described below
commit 86c8345d1e37a00c4060dc8ab43b28fda6daa243
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Thu Aug 27 11:40:43 2026 +0300
[fix][test][branch-4.2] Fix connector tests broken by the Avro 1.12.2 upgrade
Three tests that only exist on the 4.x branches (master dropped the IO
connectors in PIP-465, #25408) fail after 70998be86c2.
Avro 1.12.2 registers the JSR-310 logical type conversions on the
SpecificData.get() and ReflectData.get() singletons; earlier versions
registered none. `new SpecificDatumReader<>(schema)` uses that singleton,
so the round trip in the kinesis and elastic-search JsonConverterTest
started handing back LocalDate/Instant/UUID instead of Integer/Long/Utf8
and JsonConverter rejected them.
Pulsar itself is unaffected: GenericAvroReader decodes with a plain
GenericDatumReader, and AvroReader/AvroWriter pass an explicit
`new ReflectData()`, none of which carry the conversions. The tests were
simply round-tripping through a different decoder than the sinks see, so
they now use GenericDatumWriter/GenericDatumReader.
AvroKafkaSourceTest drives Avro's reflect API directly rather than going
through Schema.AVRO(MyBean.class), so nothing declared MyBean to the
allow-list Avro 1.12.2 enforces on reflective class resolution and
serialization failed with SecurityException. Declare it with
AvroTrustedClasses.trust(), the way an application using raw Avro would.
---
.../apache/pulsar/io/elasticsearch/JsonConverterTest.java | 15 ++++++++++-----
.../apache/pulsar/io/kinesis/json/JsonConverterTest.java | 15 ++++++++++-----
.../tests/integration/io/sources/AvroKafkaSourceTest.java | 5 +++++
3 files changed, 25 insertions(+), 10 deletions(-)
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/JsonConverterTest.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/JsonConverterTest.java
index 577bb22fcf5b..81fa9d08c57c 100644
--- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/JsonConverterTest.java
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/JsonConverterTest.java
@@ -37,14 +37,14 @@ import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
-import org.apache.avro.specific.SpecificDatumReader;
-import org.apache.avro.specific.SpecificDatumWriter;
import org.testng.annotations.Test;
public class JsonConverterTest {
@@ -148,8 +148,13 @@ public class JsonConverterTest {
assertEquals(UUID.fromString(jsonNode.get("myuuid").asText()), myUuid);
}
+ // Encodes and decodes the way Pulsar itself does when a sink consumes an Avro topic: GenericAvroReader
+ // uses a plain GenericDatumReader, so the values JsonConverter receives are the raw Avro ones
+ // (Integer, Long, Utf8) rather than converted logical-type objects. SpecificData and ReflectData must
+ // not be used here: since Avro 1.12.2 their shared singletons carry the JSR-310 conversions, so the
+ // same round trip would hand back LocalDate/Instant/UUID and would no longer test what the sink sees.
public static byte[] serialize(GenericRecord record, Schema schema) throws IOException {
- SpecificDatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<>(schema);
+ GenericDatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema);
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
BinaryEncoder binaryEncoder = new EncoderFactory().binaryEncoder(byteArrayOutputStream, null);
datumWriter.write(record, binaryEncoder);
@@ -157,8 +162,8 @@ public class JsonConverterTest {
return byteArrayOutputStream.toByteArray();
}
- public static GenericRecord deserialize(byte[] recordBytes, Schema schema) throws IOException {
- DatumReader<GenericRecord> datumReader = new SpecificDatumReader<>(schema);
+ public static GenericRecord deserialize(byte[] recordBytes, Schema schema) throws IOException {
+ DatumReader<GenericRecord> datumReader = new GenericDatumReader<>(schema);
ByteArrayInputStream stream = new ByteArrayInputStream(recordBytes);
BinaryDecoder binaryDecoder = new DecoderFactory().binaryDecoder(stream, null);
return datumReader.read(null, binaryDecoder);
diff --git a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/json/JsonConverterTest.java b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/json/JsonConverterTest.java
index 4e70b233bf11..3d3dbbda3764 100644
--- a/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/json/JsonConverterTest.java
+++ b/pulsar-io/kinesis/src/test/java/org/apache/pulsar/io/kinesis/json/JsonConverterTest.java
@@ -38,14 +38,14 @@ import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.EncoderFactory;
-import org.apache.avro.specific.SpecificDatumReader;
-import org.apache.avro.specific.SpecificDatumWriter;
import org.testng.annotations.Test;
public class JsonConverterTest {
@@ -148,8 +148,13 @@ public class JsonConverterTest {
assertEquals(UUID.fromString(jsonNode.get("myuuid").asText()), myUuid);
}
+ // Encodes and decodes the way Pulsar itself does when a sink consumes an Avro topic: GenericAvroReader
+ // uses a plain GenericDatumReader, so the values JsonConverter receives are the raw Avro ones
+ // (Integer, Long, Utf8) rather than converted logical-type objects. SpecificData and ReflectData must
+ // not be used here: since Avro 1.12.2 their shared singletons carry the JSR-310 conversions, so the
+ // same round trip would hand back LocalDate/Instant/UUID and would no longer test what the sink sees.
public static byte[] serialize(GenericRecord record, Schema schema) throws IOException {
- SpecificDatumWriter<GenericRecord> datumWriter = new SpecificDatumWriter<>(schema);
+ GenericDatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema);
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
BinaryEncoder binaryEncoder = new EncoderFactory().binaryEncoder(byteArrayOutputStream, null);
datumWriter.write(record, binaryEncoder);
@@ -157,8 +162,8 @@ public class JsonConverterTest {
return byteArrayOutputStream.toByteArray();
}
- public static GenericRecord deserialize(byte[] recordBytes, Schema schema) throws IOException {
- DatumReader<GenericRecord> datumReader = new SpecificDatumReader<>(schema);
+ public static GenericRecord deserialize(byte[] recordBytes, Schema schema) throws IOException {
+ DatumReader<GenericRecord> datumReader = new GenericDatumReader<>(schema);
ByteArrayInputStream stream = new ByteArrayInputStream(recordBytes);
BinaryDecoder binaryDecoder = new DecoderFactory().binaryDecoder(stream, null);
return datumReader.read(null, binaryDecoder);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/AvroKafkaSourceTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/AvroKafkaSourceTest.java
index b5cb3f268fce..9695805099bf 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/AvroKafkaSourceTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/AvroKafkaSourceTest.java
@@ -48,6 +48,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.schema.Field;
import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.schema.AvroTrustedClasses;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.SourceStatus;
import org.apache.pulsar.common.policies.data.SourceStatusUtil;
@@ -355,6 +356,10 @@ public class AvroKafkaSourceTest extends PulsarFunctionsTestBase {
}
public List<MyBean> produceSourceMessages(int numMessages) throws Exception{
+ // This test drives Avro's reflect API directly rather than through Schema.AVRO(MyBean.class), so
+ // nothing has declared MyBean to the allow-list Avro 1.12.2 enforces on reflective class
+ // resolution. Declare it here, the same way an application using raw Avro would.
+ AvroTrustedClasses.trust(MyBean.class);
org.apache.avro.Schema schema = ReflectData.get().getSchema(MyBean.class);
String schemaDef = schema.toString(false);
log.info("schema {}", schemaDef);