You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2019/02/28 14:35:23 UTC
[pulsar] branch master updated: [schema] Introduce `GenericSchema`
interface (#3683)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new f4d5662 [schema] Introduce `GenericSchema` interface (#3683)
f4d5662 is described below
commit f4d566248c56e5b4cfb1f74f74d3a35bdebf0e61
Author: Sijie Guo <gu...@gmail.com>
AuthorDate: Thu Feb 28 22:35:17 2019 +0800
[schema] Introduce `GenericSchema` interface (#3683)
*Motivation*
In order to introduce `GenericRecordBuilder`, we need to know the fields in a `GenericSchema`.
Otherwise, there is no way for us to build a GenericRecord.
*Modifications*
This proposal refactors current generic schema by introducing a `GenericSchema`. This generic schema
provides interfaces to retrieve the fields of a `GenericRecordSchema`.
*Additionally*
This proposal adding the primitive schemas into `Schema` class. So people can program primitive schemas
using Schema interface rather than specific implementations.
---
.../java/org/apache/pulsar/client/api/Schema.java | 53 ++++++++++++++++++
.../pulsar/client/api/schema/GenericSchema.java | 36 +++++++++++++
.../client/internal/DefaultImplementation.java | 63 ++++++++++++++++++++++
.../pulsar/client/impl/PulsarClientImpl.java | 6 +--
.../client/impl/schema/AutoConsumeSchema.java | 4 +-
.../impl/schema/generic/GenericAvroSchema.java | 2 +-
.../impl/schema/generic/GenericJsonSchema.java | 2 +-
.../{GenericSchema.java => GenericSchemaImpl.java} | 12 +++--
.../client/impl/schema/PrimitiveSchemaTest.java | 41 ++++++++------
...cSchemaTest.java => GenericSchemaImplTest.java} | 10 ++--
.../io/hbase/sink/HbaseGenericRecordSinkTest.java | 4 +-
.../org/apache/pulsar/io/jdbc/JdbcSinkTest.java | 4 +-
12 files changed, 203 insertions(+), 34 deletions(-)
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
index 8cef700..58b707e 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -18,7 +18,9 @@
*/
package org.apache.pulsar.client.api;
+import java.nio.ByteBuffer;
import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
import org.apache.pulsar.client.internal.DefaultImplementation;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.SchemaInfo;
@@ -110,11 +112,51 @@ public interface Schema<T> {
Schema<byte[]> BYTES = DefaultImplementation.newBytesSchema();
/**
+ * ByteBuffer Schema.
+ */
+ Schema<ByteBuffer> BYTEBUFFER = DefaultImplementation.newByteBufferSchema();
+
+ /**
* Schema that can be used to encode/decode messages whose values are String. The payload is encoded with UTF-8.
*/
Schema<String> STRING = DefaultImplementation.newStringSchema();
/**
+ * INT8 Schema
+ */
+ Schema<Byte> INT8 = DefaultImplementation.newByteSchema();
+
+ /**
+ * INT16 Schema
+ */
+ Schema<Short> INT16 = DefaultImplementation.newShortSchema();
+
+ /**
+ * INT32 Schema
+ */
+ Schema<Integer> INT32 = DefaultImplementation.newIntSchema();
+
+ /**
+ * INT64 Schema
+ */
+ Schema<Long> INT64 = DefaultImplementation.newLongSchema();
+
+ /**
+ * Boolean Schema
+ */
+ Schema<Boolean> BOOL = DefaultImplementation.newBooleanSchema();
+
+ /**
+ * Float Schema
+ */
+ Schema<Float> FLOAT = DefaultImplementation.newFloatSchema();
+
+ /**
+ * Double Schema
+ */
+ Schema<Double> DOUBLE = DefaultImplementation.newDoubleSchema();
+
+ /**
* Create a Protobuf schema type by extracting the fields of the specified class.
*
* @param clazz the Protobuf generated class to be used to extract the schema
@@ -208,4 +250,15 @@ public interface Schema<T> {
return DefaultImplementation.getSchema(schemaInfo);
}
+ /**
+ * Returns a generic schema of existing schema info.
+ *
+ * <p>Only supports AVRO and JSON.
+ *
+ * @param schemaInfo schema info
+ * @return a generic schema instance
+ */
+ static GenericSchema generic(SchemaInfo schemaInfo) {
+ return DefaultImplementation.getGenericSchema(schemaInfo);
+ }
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/GenericSchema.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/GenericSchema.java
new file mode 100644
index 0000000..9823c8e
--- /dev/null
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/GenericSchema.java
@@ -0,0 +1,36 @@
+/**
+ * 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.api.schema;
+
+import java.util.List;
+import org.apache.pulsar.client.api.Schema;
+
+/**
+ * A schema that serializes and deserializes between {@link GenericRecord} and bytes.
+ */
+public interface GenericSchema extends Schema<GenericRecord> {
+
+ /**
+ * Returns the list of fields.
+ *
+ * @return the list of fields of generic record.
+ */
+ List<Field> getFields();
+
+}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
index 6096e39..a4f4cfd 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/internal/DefaultImplementation.java
@@ -25,6 +25,8 @@ import static org.apache.pulsar.client.internal.ReflectionUtils.newClassInstance
import java.lang.reflect.Constructor;
import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
import java.util.Map;
import java.util.function.Supplier;
@@ -36,6 +38,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.GenericSchema;
import org.apache.pulsar.client.api.schema.RecordSchemaBuilder;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.SchemaInfo;
@@ -125,6 +128,60 @@ public class DefaultImplementation {
.newInstance());
}
+ public static Schema<String> newStringSchema(Charset charset) {
+ return catchExceptions(
+ () -> (Schema<String>) getConstructor("org.apache.pulsar.client.impl.schema.StringSchema", Charset.class)
+ .newInstance(charset));
+ }
+
+ public static Schema<Byte> newByteSchema() {
+ return catchExceptions(
+ () -> (Schema<Byte>) newClassInstance("org.apache.pulsar.client.impl.schema.ByteSchema")
+ .newInstance());
+ }
+
+ public static Schema<Short> newShortSchema() {
+ return catchExceptions(
+ () -> (Schema<Short>) newClassInstance("org.apache.pulsar.client.impl.schema.ShortSchema")
+ .newInstance());
+ }
+
+ public static Schema<Integer> newIntSchema() {
+ return catchExceptions(
+ () -> (Schema<Integer>) newClassInstance("org.apache.pulsar.client.impl.schema.IntSchema")
+ .newInstance());
+ }
+
+ public static Schema<Long> newLongSchema() {
+ return catchExceptions(
+ () -> (Schema<Long>) newClassInstance("org.apache.pulsar.client.impl.schema.LongSchema")
+ .newInstance());
+ }
+
+ public static Schema<Boolean> newBooleanSchema() {
+ return catchExceptions(
+ () -> (Schema<Boolean>) newClassInstance("org.apache.pulsar.client.impl.schema.BooleanSchema")
+ .newInstance());
+ }
+
+ public static Schema<ByteBuffer> newByteBufferSchema() {
+ return catchExceptions(
+ () -> (Schema<ByteBuffer>) newClassInstance("org.apache.pulsar.client.impl.schema.ByteBufferSchema")
+ .newInstance());
+ }
+
+ public static Schema<Float> newFloatSchema() {
+ return catchExceptions(
+ () -> (Schema<Float>) newClassInstance("org.apache.pulsar.client.impl.schema.FloatSchema")
+ .newInstance());
+ }
+
+ public static Schema<Double> newDoubleSchema() {
+ return catchExceptions(
+ () -> (Schema<Double>) newClassInstance("org.apache.pulsar.client.impl.schema.DoubleSchema")
+ .newInstance());
+ }
+
public static <T> Schema<T> newAvroSchema(Class<T> clazz) {
return catchExceptions(
() -> (Schema<T>) getStaticMethod("org.apache.pulsar.client.impl.schema.AvroSchema", "of", Class.class)
@@ -173,6 +230,12 @@ public class DefaultImplementation {
"getSchema", SchemaInfo.class).invoke(null, schemaInfo));
}
+ public static GenericSchema getGenericSchema(SchemaInfo schemaInfo) {
+ return catchExceptions(
+ () -> (GenericSchema) getStaticMethod("org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl",
+ "of", SchemaInfo.class).invoke(null, schemaInfo));
+ }
+
public static RecordSchemaBuilder newRecordSchemaBuilder(String name) {
return catchExceptions(
() -> (RecordSchemaBuilder) getConstructor("org.apache.pulsar.client.impl.schema.RecordSchemaBuilderImpl",
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 a224eb4..039f318 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
@@ -59,7 +59,7 @@ import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
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.impl.schema.generic.GenericSchemaImpl;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace.Mode;
@@ -296,7 +296,7 @@ public class PulsarClientImpl implements PulsarClient {
return lookup.getSchema(TopicName.get(conf.getSingleTopic()))
.thenCompose(schemaInfoOptional -> {
if (schemaInfoOptional.isPresent() && schemaInfoOptional.get().getType() == SchemaType.AVRO) {
- GenericSchema genericSchema = GenericSchema.of(schemaInfoOptional.get());
+ GenericSchemaImpl genericSchema = GenericSchemaImpl.of(schemaInfoOptional.get());
log.info("Auto detected schema for topic {} : {}",
conf.getSingleTopic(), new String(schemaInfoOptional.get().getSchema(), UTF_8));
autoConsumeSchema.setSchema(genericSchema);
@@ -436,7 +436,7 @@ public class PulsarClientImpl implements PulsarClient {
return lookup.getSchema(TopicName.get(conf.getTopicName()))
.thenCompose(schemaInfoOptional -> {
if (schemaInfoOptional.isPresent() && schemaInfoOptional.get().getType() == SchemaType.AVRO) {
- GenericSchema genericSchema = GenericSchema.of(schemaInfoOptional.get());
+ GenericSchemaImpl genericSchema = GenericSchemaImpl.of(schemaInfoOptional.get());
log.info("Auto detected schema for topic {} : {}",
conf.getTopicName(), new String(schemaInfoOptional.get().getSchema(), UTF_8));
autoConsumeSchema.setSchema(genericSchema);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java
index 1bd1c2c..b3443bc 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java
@@ -23,7 +23,7 @@ import static com.google.common.base.Preconditions.checkState;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.schema.GenericRecord;
-import org.apache.pulsar.client.impl.schema.generic.GenericSchema;
+import org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
import org.apache.pulsar.common.schema.SchemaInfo;
/**
@@ -89,7 +89,7 @@ public class AutoConsumeSchema implements Schema<GenericRecord> {
return BytesSchema.of();
case JSON:
case AVRO:
- return GenericSchema.of(schemaInfo);
+ return GenericSchemaImpl.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/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java
index 8cb6818..b3bcecd 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java
@@ -35,7 +35,7 @@ import org.apache.pulsar.common.schema.SchemaInfo;
/**
* A generic avro schema.
*/
-class GenericAvroSchema extends GenericSchema {
+class GenericAvroSchema extends GenericSchemaImpl {
private final GenericDatumWriter<org.apache.avro.generic.GenericRecord> datumWriter;
private BinaryEncoder encoder;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonSchema.java
index 579806f..69ebc35 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonSchema.java
@@ -31,7 +31,7 @@ import org.apache.pulsar.common.schema.SchemaInfo;
/**
* A generic json schema.
*/
-class GenericJsonSchema extends GenericSchema {
+class GenericJsonSchema extends GenericSchemaImpl {
private final ObjectMapper objectMapper;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImpl.java
similarity index 88%
rename from pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericSchema.java
rename to pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImpl.java
index 76ff3fe..69b3e15 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImpl.java
@@ -23,6 +23,7 @@ import java.util.stream.Collectors;
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.api.schema.GenericSchema;
import org.apache.pulsar.common.schema.SchemaInfo;
import static java.nio.charset.StandardCharsets.UTF_8;
@@ -30,13 +31,13 @@ import static java.nio.charset.StandardCharsets.UTF_8;
/**
* A generic schema representation.
*/
-public abstract class GenericSchema implements Schema<GenericRecord> {
+public abstract class GenericSchemaImpl implements GenericSchema {
protected final org.apache.avro.Schema schema;
protected final List<Field> fields;
protected final SchemaInfo schemaInfo;
- protected GenericSchema(SchemaInfo schemaInfo) {
+ protected GenericSchemaImpl(SchemaInfo schemaInfo) {
this.schemaInfo = schemaInfo;
this.schema = new org.apache.avro.Schema.Parser().parse(
new String(schemaInfo.getSchema(), UTF_8)
@@ -52,6 +53,11 @@ public abstract class GenericSchema implements Schema<GenericRecord> {
}
@Override
+ public List<Field> getFields() {
+ return fields;
+ }
+
+ @Override
public SchemaInfo getSchemaInfo() {
return schemaInfo;
}
@@ -62,7 +68,7 @@ public abstract class GenericSchema implements Schema<GenericRecord> {
* @param schemaInfo schema info
* @return a generic schema instance
*/
- public static GenericSchema of(SchemaInfo schemaInfo) {
+ public static GenericSchemaImpl of(SchemaInfo schemaInfo) {
switch (schemaInfo.getType()) {
case AVRO:
return new GenericAvroSchema(schemaInfo);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/PrimitiveSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/PrimitiveSchemaTest.java
index 0084997..8326b56 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/PrimitiveSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/PrimitiveSchemaTest.java
@@ -30,17 +30,8 @@ import java.util.List;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.impl.schema.ByteBufSchema;
-import org.apache.pulsar.client.impl.schema.ByteBufferSchema;
-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.LongSchema;
-import org.apache.pulsar.client.impl.schema.ShortSchema;
-import org.apache.pulsar.client.impl.schema.StringSchema;
import org.apache.pulsar.common.schema.SchemaType;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
/**
@@ -49,7 +40,7 @@ import org.testng.annotations.Test;
@Slf4j
public class PrimitiveSchemaTest {
- final private Map<Schema, List<Object>> testData = new HashMap() {
+ private static final Map<Schema, List<Object>> testData = new HashMap() {
{
put(BooleanSchema.of(), Arrays.asList(false, true));
put(StringSchema.utf8(), Arrays.asList("my string"));
@@ -65,8 +56,28 @@ public class PrimitiveSchemaTest {
}
};
- @Test
- public void allSchemasShouldSupportNull() {
+ private static final Map<Schema, List<Object>> testData2 = new HashMap() {
+ {
+ put(Schema.BOOL, Arrays.asList(false, true));
+ put(Schema.STRING, Arrays.asList("my string"));
+ put(Schema.INT8, Arrays.asList((byte) 32767, (byte) -32768));
+ put(Schema.INT16, Arrays.asList((short) 32767, (short) -32768));
+ put(Schema.INT32, Arrays.asList((int) 423412424, (int) -41243432));
+ put(Schema.INT64, Arrays.asList(922337203685477580L, -922337203685477581L));
+ put(Schema.FLOAT, Arrays.asList(5678567.12312f, -5678567.12341f));
+ put(Schema.DOUBLE, Arrays.asList(5678567.12312d, -5678567.12341d));
+ put(Schema.BYTES, Arrays.asList("my string".getBytes(UTF_8)));
+ put(Schema.BYTEBUFFER, Arrays.asList(ByteBuffer.allocate(10).put("my string".getBytes(UTF_8))));
+ }
+ };
+
+ @DataProvider(name = "schemas")
+ public Object[][] schemas() {
+ return new Object[][] { { testData }, { testData2 } };
+ }
+
+ @Test(dataProvider = "schemas")
+ public void allSchemasShouldSupportNull(Map<Schema, List<Object>> testData) {
for (Schema<?> schema : testData.keySet()) {
assertNull(schema.encode(null),
"Should support null in " + schema.getSchemaInfo().getName() + " serialization");
@@ -75,8 +86,8 @@ public class PrimitiveSchemaTest {
}
}
- @Test
- public void allSchemasShouldRoundtripInput() {
+ @Test(dataProvider = "schemas")
+ public void allSchemasShouldRoundtripInput(Map<Schema, List<Object>> testData) {
for (Map.Entry<Schema, List<Object>> test : testData.entrySet()) {
log.info("Test schema {}", test.getKey());
for (Object value : test.getValue()) {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java
similarity index 90%
rename from pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java
rename to pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java
index 4e6e261..fc65d70 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericSchemaImplTest.java
@@ -34,19 +34,19 @@ import org.testng.annotations.Test;
* Unit testing generic schemas.
*/
@Slf4j
-public class GenericSchemaTest {
+public class GenericSchemaImplTest {
@Test
public void testGenericAvroSchema() {
Schema<Foo> encodeSchema = Schema.AVRO(Foo.class);
- GenericSchema decodeSchema = GenericSchema.of(encodeSchema.getSchemaInfo());
+ GenericSchemaImpl decodeSchema = GenericSchemaImpl.of(encodeSchema.getSchemaInfo());
testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
}
@Test
public void testGenericJsonSchema() {
Schema<Foo> encodeSchema = Schema.JSON(Foo.class);
- GenericSchema decodeSchema = GenericSchema.of(encodeSchema.getSchemaInfo());
+ GenericSchemaImpl decodeSchema = GenericSchemaImpl.of(encodeSchema.getSchemaInfo());
testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
}
@@ -54,7 +54,7 @@ public class GenericSchemaTest {
public void testAutoAvroSchema() {
Schema<Foo> encodeSchema = Schema.AVRO(Foo.class);
AutoConsumeSchema decodeSchema = new AutoConsumeSchema();
- decodeSchema.setSchema(GenericSchema.of(encodeSchema.getSchemaInfo()));
+ decodeSchema.setSchema(GenericSchemaImpl.of(encodeSchema.getSchemaInfo()));
testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
}
@@ -62,7 +62,7 @@ public class GenericSchemaTest {
public void testAutoJsonSchema() {
Schema<Foo> encodeSchema = Schema.JSON(Foo.class);
AutoConsumeSchema decodeSchema = new AutoConsumeSchema();
- decodeSchema.setSchema(GenericSchema.of(encodeSchema.getSchemaInfo()));
+ decodeSchema.setSchema(GenericSchemaImpl.of(encodeSchema.getSchemaInfo()));
testEncodeAndDecodeGenericRecord(encodeSchema, decodeSchema);
}
diff --git a/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java b/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
index 2ccb08d..0478775 100644
--- a/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
+++ b/pulsar-io/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
@@ -34,7 +34,7 @@ import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.impl.MessageImpl;
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.client.impl.schema.generic.GenericSchemaImpl;
import org.apache.pulsar.common.functions.FunctionConfig;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.source.PulsarRecord;
@@ -113,7 +113,7 @@ public class HbaseGenericRecordSinkTest {
byte[] bytes = schema.encode(obj);
ByteBuf payload = Unpooled.copiedBuffer(bytes);
AutoConsumeSchema autoConsumeSchema = new AutoConsumeSchema();
- autoConsumeSchema.setSchema(GenericSchema.of(schema.getSchemaInfo()));
+ autoConsumeSchema.setSchema(GenericSchemaImpl.of(schema.getSchemaInfo()));
PulsarSourceConfig pulsarSourceConfig = new PulsarSourceConfig();
Consumer consumer = mock(Consumer.class);
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 24a7c6e..7cbaa57 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
@@ -32,7 +32,7 @@ import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.impl.MessageImpl;
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.client.impl.schema.generic.GenericSchemaImpl;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.source.PulsarRecord;
import org.testng.Assert;
@@ -100,7 +100,7 @@ public class JdbcSinkTest {
byte[] bytes = schema.encode(obj);
ByteBuf payload = Unpooled.copiedBuffer(bytes);
AutoConsumeSchema autoConsumeSchema = new AutoConsumeSchema();
- autoConsumeSchema.setSchema(GenericSchema.of(schema.getSchemaInfo()));
+ autoConsumeSchema.setSchema(GenericSchemaImpl.of(schema.getSchemaInfo()));
Message<GenericRecord> message = new MessageImpl("fake_topic_name", "77:777", conf, payload, autoConsumeSchema);
Record<GenericRecord> record = PulsarRecord.<GenericRecord>builder()