From 62c5d82fc0cc00df6c20382dafca147c984e1637 Mon Sep 17 00:00:00 2001 From: "volodymyr.burenin" Date: Tue, 2 Mar 2021 13:33:25 -0600 Subject: [PATCH 1/8] Custom avro kafka deserializer. --- .../deser/KafkaAvroSchemaDeserializer.java | 88 +++++++++++++++++++ .../utilities/sources/AvroKafkaSource.java | 24 +++-- 2 files changed, 107 insertions(+), 5 deletions(-) create mode 100644 hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java new file mode 100644 index 0000000000000..821d32fa96da5 --- /dev/null +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -0,0 +1,88 @@ +/* + * 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.hudi.utilities.deser; + +import io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer; +import io.confluent.kafka.serializers.KafkaAvroDeserializer; +import org.apache.avro.Schema; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.util.ReflectionUtils; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.utilities.schema.SchemaProvider; +import org.apache.kafka.common.errors.SerializationException; + +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; + +/** + * Extending {@link KafkaAvroSchemaDeserializer} as we need to be able to inject reader schema during deserialization. + */ +public class KafkaAvroSchemaDeserializer extends KafkaAvroDeserializer { + private static final String SCHEMA_PROVIDER_CLASS_PROP = "hoodie.deltastreamer.schemaprovider.class"; + private Schema sourceSchema; + + public KafkaAvroSchemaDeserializer() {} + + @Override + public void configure(Map configs, boolean isKey) { + super.configure(configs, isKey); + try { + TypedProperties props = getConvertToTypedProperties(configs); + String className = props.getString(SCHEMA_PROVIDER_CLASS_PROP); + SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props); + sourceSchema = Objects.requireNonNull(schemaProvider).getSourceSchema(); + } catch (Throwable e) { + throw new HoodieException(e); + } + } + + /** + * Pretty much copy-paste from the {@link AbstractKafkaAvroDeserializer} except line 87: + * DatumReader reader = new GenericDatumReader(schema, sourceSchema); + *

+ * We need to inject reader schema during deserialization or later stages of the pipeline break. + * + * @param includeSchemaAndVersion + * @param topic + * @param isKey + * @param payload + * @param readerSchema + * @return + * @throws SerializationException + */ + @Override + protected Object deserialize( + boolean includeSchemaAndVersion, + String topic, + Boolean isKey, + byte[] payload, + Schema readerSchema) + throws SerializationException { + return super.deserialize(includeSchemaAndVersion, topic, isKey, payload, sourceSchema); + } + + private TypedProperties getConvertToTypedProperties(Map configs) { + TypedProperties typedProperties = new TypedProperties(); + for (Entry entry : configs.entrySet()) { + typedProperties.put(entry.getKey(), entry.getValue()); + } + return typedProperties; + } +} diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java index 256516bd2026f..de71c849705ae 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java @@ -20,12 +20,13 @@ import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.util.Option; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.utilities.deser.KafkaAvroSchemaDeserializer; import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics; import org.apache.hudi.utilities.schema.SchemaProvider; import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen; import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.CheckpointUtils; -import io.confluent.kafka.serializers.KafkaAvroDeserializer; import org.apache.avro.generic.GenericRecord; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.log4j.LogManager; @@ -42,18 +43,31 @@ */ public class AvroKafkaSource extends AvroSource { + private static final String KAFKA_AVRO_VALUE_DESERIALIZER = "hoodie.deltastreamer.source.kafka.value.deserializer.class"; private static final Logger LOG = LogManager.getLogger(AvroKafkaSource.class); - private final KafkaOffsetGen offsetGen; - private final HoodieDeltaStreamerMetrics metrics; public AvroKafkaSource(TypedProperties props, JavaSparkContext sparkContext, SparkSession sparkSession, SchemaProvider schemaProvider, HoodieDeltaStreamerMetrics metrics) { super(props, sparkContext, sparkSession, schemaProvider); - this.metrics = metrics; + props.put("key.deserializer", StringDeserializer.class); - props.put("value.deserializer", KafkaAvroDeserializer.class); + String deserializerClassName = props.getString(KAFKA_AVRO_VALUE_DESERIALIZER, ""); + + if (deserializerClassName.isEmpty()) { + props.put("value.deserializer", KafkaAvroSchemaDeserializer.class); + } else { + try { + props.put("value.deserializer", Class.forName(deserializerClassName)); + } catch (ClassNotFoundException e) { + String error = "Could not load custom avro kafka deserializer: " + deserializerClassName; + LOG.error(error); + throw new HoodieException(error, e); + } + } + + this.metrics = metrics; offsetGen = new KafkaOffsetGen(props); } From d3e28723a2fc691115c7c91ce4afdb10d46cfd04 Mon Sep 17 00:00:00 2001 From: "volodymyr.burenin" Date: Tue, 2 Mar 2021 13:46:53 -0600 Subject: [PATCH 2/8] Updated method comment in KafkaAvroSchemaDeserialize to reflect what it does. --- .../hudi/utilities/deser/KafkaAvroSchemaDeserializer.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java index 821d32fa96da5..9a3eba1545847 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -54,10 +54,7 @@ public void configure(Map configs, boolean isKey) { } /** - * Pretty much copy-paste from the {@link AbstractKafkaAvroDeserializer} except line 87: - * DatumReader reader = new GenericDatumReader(schema, sourceSchema); - *

- * We need to inject reader schema during deserialization or later stages of the pipeline break. + * We need to inject sourceSchema instead of reader schema during deserialization or later stages of the pipeline. * * @param includeSchemaAndVersion * @param topic From 118145765416465796f535c45739ea04b9f3b6c7 Mon Sep 17 00:00:00 2001 From: "volodymyr.burenin" Date: Tue, 2 Mar 2021 13:50:19 -0600 Subject: [PATCH 3/8] Added additional SchemaProvider constructor that uses null for JavaSparkContext. --- .../utilities/schema/NullTargetSchemaRegistryProvider.java | 3 +++ .../apache/hudi/utilities/schema/SchemaRegistryProvider.java | 4 ++++ 2 files changed, 7 insertions(+) diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java index 5983238e89a6d..4013c630c7459 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java @@ -32,6 +32,9 @@ public class NullTargetSchemaRegistryProvider extends SchemaRegistryProvider { public NullTargetSchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) { super(props, jssc); } + public NullTargetSchemaRegistryProvider(TypedProperties props) { + super(props); + } @Override public Schema getTargetSchema() { diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java index 47c4c2f81a790..da92f22bec37f 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java @@ -55,6 +55,10 @@ private static String fetchSchemaFromRegistry(String registryUrl) throws IOExcep return node.get("schema").asText(); } + public SchemaRegistryProvider(TypedProperties props) { + this(props, null); + } + public SchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) { super(props, jssc); DataSourceUtils.checkRequiredProperties(props, Collections.singletonList(Config.SRC_SCHEMA_REGISTRY_URL_PROP)); From 6a9a7f7230589e1a685d75ee1cf51120ed2a388e Mon Sep 17 00:00:00 2001 From: "volodymyr.burenin" Date: Tue, 2 Mar 2021 14:54:28 -0600 Subject: [PATCH 4/8] fixed style issues --- .../apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java | 1 - .../hudi/utilities/schema/NullTargetSchemaRegistryProvider.java | 1 + 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java index 9a3eba1545847..1b1cb56bb9864 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -18,7 +18,6 @@ package org.apache.hudi.utilities.deser; -import io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer; import io.confluent.kafka.serializers.KafkaAvroDeserializer; import org.apache.avro.Schema; import org.apache.hudi.common.config.TypedProperties; diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java index 4013c630c7459..5549c218f0045 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java @@ -32,6 +32,7 @@ public class NullTargetSchemaRegistryProvider extends SchemaRegistryProvider { public NullTargetSchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) { super(props, jssc); } + public NullTargetSchemaRegistryProvider(TypedProperties props) { super(props); } From de50b688e5b80b138d6a11ab1ffde170d0778770 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Mon, 8 Mar 2021 23:36:15 -0500 Subject: [PATCH 5/8] Adding avroSource test --- .../testutils/HoodieTestDataGenerator.java | 40 +++- .../deser/KafkaAvroSchemaDeserializer.java | 4 +- .../sources/TestAvroKafkaSource.java | 187 ++++++++++++++++++ .../KafkaAvroTestCustomDeserializer.java | 79 ++++++++ .../helpers/KafkaAvroTestDeserializer.java | 71 +++++++ .../helpers/KafkaAvroTestSerializer.java | 73 +++++++ .../sources/helpers/SchemaTestProvider.java | 43 ++++ 7 files changed, 492 insertions(+), 5 deletions(-) create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java index 8017bc3d74860..6ac7eeaa386fb 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java @@ -114,6 +114,8 @@ public class HoodieTestDataGenerator { TRIP_SCHEMA_PREFIX + EXTRA_TYPE_SCHEMA + MAP_TYPE_SCHEMA + FARE_NESTED_SCHEMA + TIP_NESTED_SCHEMA + TRIP_SCHEMA_SUFFIX; public static final String TRIP_FLATTENED_SCHEMA = TRIP_SCHEMA_PREFIX + FARE_FLATTENED_SCHEMA + TRIP_SCHEMA_SUFFIX; + public static final String TRIP_EVOLVED_EXAMPLE_SCHEMA = TRIP_EXAMPLE_SCHEMA.substring(0, TRIP_EXAMPLE_SCHEMA.length() - 2) + + ",{\"name\":\"rider_evolved\",\"type\":\"string\"}]}"; public static final String TRIP_SCHEMA = "{\"type\":\"record\",\"name\":\"tripUberRec\",\"fields\":[" + "{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"}," @@ -126,8 +128,8 @@ public class HoodieTestDataGenerator { public static final String TRIP_HIVE_COLUMN_TYPES = "bigint,string,string,string,double,double,double,double,int,bigint,float,binary,int,bigint,decimal(10,6)," + "map,struct,array>,boolean"; - public static final Schema AVRO_SCHEMA = new Schema.Parser().parse(TRIP_EXAMPLE_SCHEMA); + public static final Schema AVRO_EVOLVED_SCHEMA = new Schema.Parser().parse(TRIP_EVOLVED_EXAMPLE_SCHEMA); public static final Schema AVRO_SCHEMA_WITH_METADATA_FIELDS = HoodieAvroUtils.addMetadataFields(AVRO_SCHEMA); public static final Schema AVRO_SHORT_TRIP_SCHEMA = new Schema.Parser().parse(SHORT_TRIP_SCHEMA); @@ -178,6 +180,8 @@ public RawTripTestPayload generateRandomValueAsPerSchema(String schemaStr, Hoodi return generatePayloadForTripSchema(key, commitTime); } else if (SHORT_TRIP_SCHEMA.equals(schemaStr)) { return generatePayloadForShortTripSchema(key, commitTime); + } else if (TRIP_EVOLVED_EXAMPLE_SCHEMA.equals(schemaStr)) { + return generateRandomValueForEvolvedSchema(key, commitTime, isFlattened); } return null; @@ -213,6 +217,15 @@ public static RawTripTestPayload generateRandomValue( return new RawTripTestPayload(rec.toString(), key.getRecordKey(), key.getPartitionPath(), TRIP_EXAMPLE_SCHEMA); } + public static RawTripTestPayload generateRandomValueForEvolvedSchema( + HoodieKey key, String instantTime, boolean isFlattened) throws IOException { + GenericRecord rec = generateGenericRecord( + key.getRecordKey(), "rider-" + instantTime, "driver-" + instantTime, 0, + false, isFlattened, true); + rec.put("rider_evolved", "rider-" + instantTime + "_evolved"); + return new RawTripTestPayload(rec.toString(), key.getRecordKey(), key.getPartitionPath(), TRIP_EVOLVED_EXAMPLE_SCHEMA); + } + /** * Generates a new avro record with TRIP_SCHEMA, retaining the key if optionally provided. */ @@ -248,10 +261,16 @@ public static GenericRecord generateGenericRecord(String rowKey, String riderNam return generateGenericRecord(rowKey, riderName, driverName, timestamp, false, false); } + public static GenericRecord generateGenericRecord(String rowKey, String riderName, String driverName, + long timestamp, boolean isDeleteRecord, + boolean isFlattened) { + return generateGenericRecord(rowKey, riderName, driverName, timestamp, isDeleteRecord, isFlattened, false); + } + public static GenericRecord generateGenericRecord(String rowKey, String riderName, String driverName, long timestamp, boolean isDeleteRecord, - boolean isFlattened) { - GenericRecord rec = new GenericData.Record(isFlattened ? FLATTENED_AVRO_SCHEMA : AVRO_SCHEMA); + boolean isFlattened, boolean isEvolvedSchema) { + GenericRecord rec = new GenericData.Record(isEvolvedSchema ? AVRO_EVOLVED_SCHEMA : (isFlattened ? FLATTENED_AVRO_SCHEMA : AVRO_SCHEMA)); rec.put("_row_key", rowKey); rec.put("timestamp", timestamp); rec.put("rider", riderName); @@ -318,6 +337,21 @@ public GenericRecord generateRecordForTripSchema(String rowKey, String riderName return rec; } + /* + * Generate random record using TRIP_EVOLVED_SCHEMA + */ + public GenericRecord generateRecordForEvolvedTripSchema(String rowKey, String riderName, String driverName, long timestamp) { + GenericRecord rec = new GenericData.Record(AVRO_EVOLVED_SCHEMA); + rec.put("_row_key", rowKey); + rec.put("timestamp", timestamp); + rec.put("rider", riderName); + rec.put("driver", driverName); + rec.put("fare", RAND.nextDouble() * 100); + rec.put("_hoodie_is_deleted", false); + rec.put("rider_evolved", riderName + "_evolved"); + return rec; + } + public GenericRecord generateRecordForShortTripSchema(String rowKey, String riderName, String driverName, long timestamp) { GenericRecord rec = new GenericData.Record(AVRO_SHORT_TRIP_SCHEMA); rec.put("_row_key", rowKey); diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java index 1b1cb56bb9864..27d3885580f17 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -35,7 +35,7 @@ */ public class KafkaAvroSchemaDeserializer extends KafkaAvroDeserializer { private static final String SCHEMA_PROVIDER_CLASS_PROP = "hoodie.deltastreamer.schemaprovider.class"; - private Schema sourceSchema; + protected Schema sourceSchema; public KafkaAvroSchemaDeserializer() {} @@ -74,7 +74,7 @@ protected Object deserialize( return super.deserialize(includeSchemaAndVersion, topic, isKey, payload, sourceSchema); } - private TypedProperties getConvertToTypedProperties(Map configs) { + protected TypedProperties getConvertToTypedProperties(Map configs) { TypedProperties typedProperties = new TypedProperties(); for (Entry entry : configs.entrySet()) { typedProperties.put(entry.getKey(), entry.getValue()); diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java new file mode 100644 index 0000000000000..33e7d17b140b4 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java @@ -0,0 +1,187 @@ +/* + * 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.hudi.utilities.sources; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.testutils.HoodieTestDataGenerator; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics; +import org.apache.hudi.utilities.deltastreamer.SourceFormatAdapter; +import org.apache.hudi.utilities.schema.FilebasedSchemaProvider; +import org.apache.hudi.utilities.sources.helpers.KafkaAvroTestDeserializer; +import org.apache.hudi.utilities.sources.helpers.KafkaAvroTestSerializer; +import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.Config; +import org.apache.hudi.utilities.sources.helpers.SchemaTestProvider; +import org.apache.hudi.utilities.testutils.UtilitiesTestBase; + +import org.apache.avro.generic.GenericRecord; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.spark.api.java.JavaRDD; +import org.apache.spark.streaming.kafka010.KafkaTestUtils; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import java.util.List; +import java.util.UUID; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Unit tests {@link AvroKakfaSource}. + */ +public class TestAvroKafkaSource extends UtilitiesTestBase { + + private static String TEST_TOPIC_NAME = "hoodie_test"; + + private FilebasedSchemaProvider schemaProvider; + private KafkaTestUtils testUtils; + private Producer kafkaProducer; + private HoodieDeltaStreamerMetrics metrics = mock(HoodieDeltaStreamerMetrics.class); + @Mock + private SchemaTestProvider schemaTestProvider; + + @BeforeAll + public static void initClass() throws Exception { + UtilitiesTestBase.initClass(); + } + + @AfterAll + public static void cleanupClass() { + UtilitiesTestBase.cleanupClass(); + } + + @BeforeEach + public void setup() throws Exception { + MockitoAnnotations.initMocks(this); + super.setup(); + schemaProvider = new FilebasedSchemaProvider(Helpers.setupSchemaOnDFS(), jsc); + testUtils = new KafkaTestUtils(); + testUtils.setup(); + kafkaProducer = createProducer(testUtils.brokerAddress(), "dummy"); + } + + @AfterEach + public void teardown() throws Exception { + super.teardown(); + testUtils.teardown(); + } + + private TypedProperties createPropsForAvroKafkaSource(Long maxEventsToReadFromKafkaSource, String resetStrategy) { + TypedProperties props = new TypedProperties(); + props.setProperty("hoodie.deltastreamer.source.kafka.topic", TEST_TOPIC_NAME); + props.setProperty("bootstrap.servers", testUtils.brokerAddress()); + props.setProperty("hoodie.deltastreamer.source.kafka.auto.reset.offsets", resetStrategy); + props.setProperty("hoodie.deltastreamer.kafka.source.maxEvents", + maxEventsToReadFromKafkaSource != null ? String.valueOf(maxEventsToReadFromKafkaSource) : + String.valueOf(Config.maxEventsFromKafkaSource)); + props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString()); + return props; + } + + public static Producer createProducer(String brokers, String schemaRegistryUrl) { + TypedProperties props = new TypedProperties(); + props.put("bootstrap.servers", brokers); + props.put("key.serializer", StringSerializer.class); + props.put("value.serializer", KafkaAvroTestSerializer.class); + props.put("schema.registry.url", schemaRegistryUrl); + return new KafkaProducer(props); + } + + @Test + public void testAvroKafkaSource() { + testUtils.createTopic(TEST_TOPIC_NAME, 2); + HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator(); + TypedProperties props = createPropsForAvroKafkaSource(null, "earliest"); + props.put("value.deserializer", KafkaAvroTestDeserializer.class); + props.put("hoodie.deltastreamer.source.kafka.value.deserializer.class", KafkaAvroTestDeserializer.class.getCanonicalName()); + props.put("hoodie.deltastreamer.schemaprovider.class", SchemaTestProvider.class); + when(schemaTestProvider.getSourceSchema()).thenReturn(HoodieTestDataGenerator.AVRO_SCHEMA); + + Source avroKafkaSource = new AvroKafkaSource(props, jsc, sparkSession, schemaProvider, metrics); + SourceFormatAdapter kafkaSource = new SourceFormatAdapter(avroKafkaSource); + + // 1. Extract without any checkpoint => get all the data, respecting sourceLimit + assertEquals(Option.empty(), kafkaSource.fetchNewDataInAvroFormat(Option.empty(), Long.MAX_VALUE).getBatch()); + List genericRecordList = Helpers.toGenericRecords(dataGenerator.generateInserts("000", 10)); + for (GenericRecord genericRecord : genericRecordList) { + kafkaProducer.send(new ProducerRecord<>(TEST_TOPIC_NAME, genericRecord)); + } + kafkaProducer.flush(); + kafkaProducer.close(); + + InputBatch> fetch1 = kafkaSource.fetchNewDataInAvroFormat(Option.empty(), 10); + List actualGenRecs = fetch1.getBatch().get().collect(); + assertEquals(10, actualGenRecs.size()); + // to be fixed : equality checks for generic Records. + // assertEquals(genericRecordList, actualGenRecs); + assertListEquality(genericRecordList, actualGenRecs); + + // evolve schema and add more records to kafka + KafkaAvroTestSerializer.schemaToReturn = HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA; + KafkaAvroTestDeserializer.schemaToReturn = HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA; + when(schemaTestProvider.getSourceSchema()).thenReturn(HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA); + kafkaProducer = createProducer(testUtils.brokerAddress(), "dummy"); + + genericRecordList = Helpers.toGenericRecords(dataGenerator.generateInsertsStream("001", 20, false, + HoodieTestDataGenerator.TRIP_EVOLVED_EXAMPLE_SCHEMA).collect(Collectors.toList()), HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA); + for (GenericRecord genericRecord : genericRecordList) { + kafkaProducer.send(new ProducerRecord<>(TEST_TOPIC_NAME, genericRecord)); + } + kafkaProducer.flush(); + + InputBatch> fetch2 = kafkaSource.fetchNewDataInAvroFormat(Option.of(fetch1.getCheckpointForNextBatch()), 20); + List actualGenRecs2 = fetch2.getBatch().get().collect(); + assertEquals(20, actualGenRecs2.size()); + assertEquals(genericRecordList, actualGenRecs2); + } + + private void assertListEquality(List list1, List list2) { + System.out.println("Expected :: "); + for (GenericRecord gRec : list1) { + System.out.println("exp rec:: " + gRec.toString()); + } + System.out.println("Actual :: "); + for (GenericRecord gRec : list2) { + System.out.println("act rec:: " + gRec.toString()); + } + assertEquals(list1.size(), list2.size()); + for (GenericRecord genericRecord : list1) { + System.out.println("Checking " + genericRecord.toString() + " from exp "); + assertTrue(list2.contains(genericRecord)); + } + for (GenericRecord genericRecord : list2) { + System.out.println("Checking " + genericRecord.toString() + " from actual "); + assertTrue(list1.contains(genericRecord)); + } + } + +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java new file mode 100644 index 0000000000000..dafe92aeec375 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java @@ -0,0 +1,79 @@ +/* + * 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.hudi.utilities.sources.helpers; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.testutils.HoodieTestDataGenerator; +import org.apache.hudi.common.util.ReflectionUtils; +import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.utilities.deser.KafkaAvroSchemaDeserializer; +import org.apache.hudi.utilities.schema.SchemaProvider; + +import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; +import org.apache.avro.Schema; + +import java.io.IOException; +import java.util.Map; +import java.util.Objects; + +/** + * KafkaAvroSchemaDeserializer for tests using {@link MockSchemaRegistryClient}. Since in tests we can't use schema registry, + * have to override calls especially super.configure(). So, had to introduce these test classes. Impl of configure() should + * be in line with {@link KafkaAvroSchemaDeserializer} except for super.configure(). + */ +public class KafkaAvroTestCustomDeserializer extends KafkaAvroSchemaDeserializer { + + public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; + + public KafkaAvroTestCustomDeserializer() { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + } + + public KafkaAvroTestCustomDeserializer(SchemaRegistryClient client) { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + } + + public KafkaAvroTestCustomDeserializer(SchemaRegistryClient client, Map props) { + this.schemaRegistry = client; + this.configure(this.deserializerConfig(props)); + } + + public void configure(Map configs, boolean isKey) { + try { + TypedProperties props = getConvertToTypedProperties(configs); + String className = props.getString("hoodie.deltastreamer.schemaprovider.class"); + SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props); + sourceSchema = Objects.requireNonNull(schemaProvider).getSourceSchema(); + } catch (Throwable e) { + throw new HoodieException(e); + } + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java new file mode 100644 index 0000000000000..15fbc66042c7e --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java @@ -0,0 +1,71 @@ +/* + * 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.hudi.utilities.sources.helpers; + +import org.apache.hudi.common.testutils.HoodieTestDataGenerator; + +import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; +import io.confluent.kafka.serializers.KafkaAvroDeserializer; +import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig; +import org.apache.avro.Schema; + +import java.io.IOException; +import java.util.Map; + +/** + * KafkaAvroDeserializer for tests using {@link MockSchemaRegistryClient}. Since in tests we can't use schema registry, have to override calls especially + * super.configure(). So, had to introduce these test classes. Impl of configure() should be in line with + * {@link KafkaAvroDeserializer} except for super.configure(). + */ +public class KafkaAvroTestDeserializer extends KafkaAvroDeserializer { + + public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; + + public KafkaAvroTestDeserializer() { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + } + + public KafkaAvroTestDeserializer(SchemaRegistryClient client) { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + } + + public KafkaAvroTestDeserializer(SchemaRegistryClient client, Map props) { + this.schemaRegistry = client; + this.configure(this.deserializerConfig(props)); + } + + protected void configure(KafkaAvroDeserializerConfig config) { + // no op + } + + public void configure(Map configs, boolean isKey) { + } +} \ No newline at end of file diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java new file mode 100644 index 0000000000000..59a628b5f0097 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java @@ -0,0 +1,73 @@ +/* + * 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.hudi.utilities.sources.helpers; + +import org.apache.hudi.common.testutils.HoodieTestDataGenerator; + +import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; +import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig; +import io.confluent.kafka.serializers.KafkaAvroSerializer; +import org.apache.avro.Schema; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +/** + * KafkaAvroSerializer for tests using {@link MockSchemaRegistryClient}. Since in tests we can't use schema registry, + * have to override calls especially super.configure(). So, had to introduce these test classes. Impl of configure() + * should be in line with {@link KafkaAvroSerializer} except for super.configure(). + */ +public class KafkaAvroTestSerializer extends KafkaAvroSerializer { + + public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; + + public KafkaAvroTestSerializer() { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + } + + public KafkaAvroTestSerializer(SchemaRegistryClient client) { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + } + + public KafkaAvroTestSerializer(SchemaRegistryClient client, Map props) { + this.schemaRegistry = new MockSchemaRegistryClient() { + @Override + public synchronized Schema getByID(int id) throws IOException, RestClientException { + return schemaToReturn; + } + }; + Map map = new HashMap<>(); + map.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "dummy"); + this.configure(this.serializerConfig(props)); + } + +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java new file mode 100644 index 0000000000000..b9ee464da0433 --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java @@ -0,0 +1,43 @@ +/* + * 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.hudi.utilities.sources.helpers; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.testutils.HoodieTestDataGenerator; +import org.apache.hudi.utilities.schema.SchemaProvider; + +import org.apache.avro.Schema; +import org.apache.spark.api.java.JavaSparkContext; + +/** + * {@link SchemaProvider} for tests. + */ +public class SchemaTestProvider extends SchemaProvider { + + public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; + + protected SchemaTestProvider(TypedProperties props, JavaSparkContext jssc) { + super(props, jssc); + } + + @Override + public Schema getSourceSchema() { + return schemaToReturn; + } +} From 15c058a2cfbe64c902db9bfed50bee0acf0e1403 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Sun, 14 Mar 2021 13:48:31 -0400 Subject: [PATCH 6/8] Removing unwanted method in HoodieTestDataGenerator --- .../common/testutils/HoodieTestDataGenerator.java | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java index 6ac7eeaa386fb..bf3a677c6364f 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java @@ -337,21 +337,6 @@ public GenericRecord generateRecordForTripSchema(String rowKey, String riderName return rec; } - /* - * Generate random record using TRIP_EVOLVED_SCHEMA - */ - public GenericRecord generateRecordForEvolvedTripSchema(String rowKey, String riderName, String driverName, long timestamp) { - GenericRecord rec = new GenericData.Record(AVRO_EVOLVED_SCHEMA); - rec.put("_row_key", rowKey); - rec.put("timestamp", timestamp); - rec.put("rider", riderName); - rec.put("driver", driverName); - rec.put("fare", RAND.nextDouble() * 100); - rec.put("_hoodie_is_deleted", false); - rec.put("rider_evolved", riderName + "_evolved"); - return rec; - } - public GenericRecord generateRecordForShortTripSchema(String rowKey, String riderName, String driverName, long timestamp) { GenericRecord rec = new GenericData.Record(AVRO_SHORT_TRIP_SCHEMA); rec.put("_row_key", rowKey); From bbbf899480e9d458b6cde3976ba8078d0517b752 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Thu, 18 Mar 2021 13:40:23 -0400 Subject: [PATCH 7/8] Fixing tests for kakfaSchemaDeser --- .../testutils/HoodieTestDataGenerator.java | 25 +-- .../org/apache/hudi/DataSourceOptions.scala | 8 + .../deser/KafkaAvroSchemaDeserializer.java | 17 +- .../NullTargetSchemaRegistryProvider.java | 4 - .../schema/SchemaRegistryProvider.java | 4 - .../utilities/sources/AvroKafkaSource.java | 14 +- .../TestKafkaAvroSchemaDeserializer.java | 143 ++++++++++++++ .../sources/TestAvroKafkaSource.java | 187 ------------------ .../KafkaAvroTestCustomDeserializer.java | 79 -------- .../helpers/KafkaAvroTestDeserializer.java | 71 ------- .../helpers/KafkaAvroTestSerializer.java | 73 ------- .../sources/helpers/SchemaTestProvider.java | 11 +- 12 files changed, 184 insertions(+), 452 deletions(-) create mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java delete mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java delete mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java delete mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java delete mode 100644 hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java index bf3a677c6364f..8017bc3d74860 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java @@ -114,8 +114,6 @@ public class HoodieTestDataGenerator { TRIP_SCHEMA_PREFIX + EXTRA_TYPE_SCHEMA + MAP_TYPE_SCHEMA + FARE_NESTED_SCHEMA + TIP_NESTED_SCHEMA + TRIP_SCHEMA_SUFFIX; public static final String TRIP_FLATTENED_SCHEMA = TRIP_SCHEMA_PREFIX + FARE_FLATTENED_SCHEMA + TRIP_SCHEMA_SUFFIX; - public static final String TRIP_EVOLVED_EXAMPLE_SCHEMA = TRIP_EXAMPLE_SCHEMA.substring(0, TRIP_EXAMPLE_SCHEMA.length() - 2) - + ",{\"name\":\"rider_evolved\",\"type\":\"string\"}]}"; public static final String TRIP_SCHEMA = "{\"type\":\"record\",\"name\":\"tripUberRec\",\"fields\":[" + "{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"}," @@ -128,8 +126,8 @@ public class HoodieTestDataGenerator { public static final String TRIP_HIVE_COLUMN_TYPES = "bigint,string,string,string,double,double,double,double,int,bigint,float,binary,int,bigint,decimal(10,6)," + "map,struct,array>,boolean"; + public static final Schema AVRO_SCHEMA = new Schema.Parser().parse(TRIP_EXAMPLE_SCHEMA); - public static final Schema AVRO_EVOLVED_SCHEMA = new Schema.Parser().parse(TRIP_EVOLVED_EXAMPLE_SCHEMA); public static final Schema AVRO_SCHEMA_WITH_METADATA_FIELDS = HoodieAvroUtils.addMetadataFields(AVRO_SCHEMA); public static final Schema AVRO_SHORT_TRIP_SCHEMA = new Schema.Parser().parse(SHORT_TRIP_SCHEMA); @@ -180,8 +178,6 @@ public RawTripTestPayload generateRandomValueAsPerSchema(String schemaStr, Hoodi return generatePayloadForTripSchema(key, commitTime); } else if (SHORT_TRIP_SCHEMA.equals(schemaStr)) { return generatePayloadForShortTripSchema(key, commitTime); - } else if (TRIP_EVOLVED_EXAMPLE_SCHEMA.equals(schemaStr)) { - return generateRandomValueForEvolvedSchema(key, commitTime, isFlattened); } return null; @@ -217,15 +213,6 @@ public static RawTripTestPayload generateRandomValue( return new RawTripTestPayload(rec.toString(), key.getRecordKey(), key.getPartitionPath(), TRIP_EXAMPLE_SCHEMA); } - public static RawTripTestPayload generateRandomValueForEvolvedSchema( - HoodieKey key, String instantTime, boolean isFlattened) throws IOException { - GenericRecord rec = generateGenericRecord( - key.getRecordKey(), "rider-" + instantTime, "driver-" + instantTime, 0, - false, isFlattened, true); - rec.put("rider_evolved", "rider-" + instantTime + "_evolved"); - return new RawTripTestPayload(rec.toString(), key.getRecordKey(), key.getPartitionPath(), TRIP_EVOLVED_EXAMPLE_SCHEMA); - } - /** * Generates a new avro record with TRIP_SCHEMA, retaining the key if optionally provided. */ @@ -261,16 +248,10 @@ public static GenericRecord generateGenericRecord(String rowKey, String riderNam return generateGenericRecord(rowKey, riderName, driverName, timestamp, false, false); } - public static GenericRecord generateGenericRecord(String rowKey, String riderName, String driverName, - long timestamp, boolean isDeleteRecord, - boolean isFlattened) { - return generateGenericRecord(rowKey, riderName, driverName, timestamp, isDeleteRecord, isFlattened, false); - } - public static GenericRecord generateGenericRecord(String rowKey, String riderName, String driverName, long timestamp, boolean isDeleteRecord, - boolean isFlattened, boolean isEvolvedSchema) { - GenericRecord rec = new GenericData.Record(isEvolvedSchema ? AVRO_EVOLVED_SCHEMA : (isFlattened ? FLATTENED_AVRO_SCHEMA : AVRO_SCHEMA)); + boolean isFlattened) { + GenericRecord rec = new GenericData.Record(isFlattened ? FLATTENED_AVRO_SCHEMA : AVRO_SCHEMA); rec.put("_row_key", rowKey); rec.put("timestamp", timestamp); rec.put("rider", riderName); diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala index 4b8e97cf53df2..42e303c58b266 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala @@ -373,4 +373,12 @@ object DataSourceWriteOptions { // Async Compaction - Enabled by default for MOR val ASYNC_COMPACT_ENABLE_OPT_KEY = "hoodie.datasource.compaction.async.enable" val DEFAULT_ASYNC_COMPACT_ENABLE_OPT_VAL = "true" + + // Avro Kafka Source configs + val KAFKA_AVRO_VALUE_DESERIALIZER = "hoodie.deltastreamer.source.kafka.value.deserializer.class" + // val DEFAULT_KAFKA_AVRO_VALUE_DESERIALIZER = classOf[io.confluent.kafka.serializers.KafkaAvroDeserializer] + + val SCHEMA_PROVIDER_CLASS_PROP = "hoodie.deltastreamer.schemaprovider.class" + + val JAVA_SPARK_CONTEXT_PROP = "java.spark.context" } diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java index 27d3885580f17..9d429de834409 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -18,13 +18,17 @@ package org.apache.hudi.utilities.deser; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; import io.confluent.kafka.serializers.KafkaAvroDeserializer; import org.apache.avro.Schema; + +import org.apache.hudi.DataSourceWriteOptions; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.util.ReflectionUtils; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.utilities.schema.SchemaProvider; import org.apache.kafka.common.errors.SerializationException; +import org.apache.spark.api.java.JavaSparkContext; import java.util.Map; import java.util.Map.Entry; @@ -34,18 +38,23 @@ * Extending {@link KafkaAvroSchemaDeserializer} as we need to be able to inject reader schema during deserialization. */ public class KafkaAvroSchemaDeserializer extends KafkaAvroDeserializer { - private static final String SCHEMA_PROVIDER_CLASS_PROP = "hoodie.deltastreamer.schemaprovider.class"; - protected Schema sourceSchema; + + private Schema sourceSchema; public KafkaAvroSchemaDeserializer() {} + public KafkaAvroSchemaDeserializer(SchemaRegistryClient client, Map props) { + super(client, props); + } + @Override public void configure(Map configs, boolean isKey) { super.configure(configs, isKey); try { TypedProperties props = getConvertToTypedProperties(configs); - String className = props.getString(SCHEMA_PROVIDER_CLASS_PROP); - SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props); + String className = props.getString(DataSourceWriteOptions.SCHEMA_PROVIDER_CLASS_PROP()); + JavaSparkContext jsc = (JavaSparkContext) props.get(DataSourceWriteOptions.JAVA_SPARK_CONTEXT_PROP()); + SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props, jsc); sourceSchema = Objects.requireNonNull(schemaProvider).getSourceSchema(); } catch (Throwable e) { throw new HoodieException(e); diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java index 5549c218f0045..5983238e89a6d 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/NullTargetSchemaRegistryProvider.java @@ -32,10 +32,6 @@ public class NullTargetSchemaRegistryProvider extends SchemaRegistryProvider { public NullTargetSchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) { super(props, jssc); } - - public NullTargetSchemaRegistryProvider(TypedProperties props) { - super(props); - } @Override public Schema getTargetSchema() { diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java index da92f22bec37f..47c4c2f81a790 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java @@ -55,10 +55,6 @@ private static String fetchSchemaFromRegistry(String registryUrl) throws IOExcep return node.get("schema").asText(); } - public SchemaRegistryProvider(TypedProperties props) { - this(props, null); - } - public SchemaRegistryProvider(TypedProperties props, JavaSparkContext jssc) { super(props, jssc); DataSourceUtils.checkRequiredProperties(props, Collections.singletonList(Config.SRC_SCHEMA_REGISTRY_URL_PROP)); diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java index de71c849705ae..bc7c42cf1e438 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java @@ -18,15 +18,17 @@ package org.apache.hudi.utilities.sources; +import org.apache.hudi.DataSourceWriteOptions; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.util.Option; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.utilities.deser.KafkaAvroSchemaDeserializer; +import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics; import org.apache.hudi.utilities.schema.SchemaProvider; import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen; import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.CheckpointUtils; +import io.confluent.kafka.serializers.KafkaAvroDeserializer; import org.apache.avro.generic.GenericRecord; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.log4j.LogManager; @@ -43,7 +45,6 @@ */ public class AvroKafkaSource extends AvroSource { - private static final String KAFKA_AVRO_VALUE_DESERIALIZER = "hoodie.deltastreamer.source.kafka.value.deserializer.class"; private static final Logger LOG = LogManager.getLogger(AvroKafkaSource.class); private final KafkaOffsetGen offsetGen; private final HoodieDeltaStreamerMetrics metrics; @@ -53,12 +54,17 @@ public AvroKafkaSource(TypedProperties props, JavaSparkContext sparkContext, Spa super(props, sparkContext, sparkSession, schemaProvider); props.put("key.deserializer", StringDeserializer.class); - String deserializerClassName = props.getString(KAFKA_AVRO_VALUE_DESERIALIZER, ""); + String deserializerClassName = props.getString(DataSourceWriteOptions.KAFKA_AVRO_VALUE_DESERIALIZER(), ""); if (deserializerClassName.isEmpty()) { - props.put("value.deserializer", KafkaAvroSchemaDeserializer.class); + props.put("value.deserializer", KafkaAvroDeserializer.class); } else { try { + if (schemaProvider == null) { + throw new HoodieIOException("SchemaProvider has to be set to use custom Deserializer"); + } + props.put(DataSourceWriteOptions.SCHEMA_PROVIDER_CLASS_PROP(), schemaProvider.getClass().getName()); + props.put(DataSourceWriteOptions.JAVA_SPARK_CONTEXT_PROP(), sparkContext); props.put("value.deserializer", Class.forName(deserializerClassName)); } catch (ClassNotFoundException e) { String error = "Could not load custom avro kafka deserializer: " + deserializerClassName; diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java new file mode 100644 index 0000000000000..beee0d3c3c49e --- /dev/null +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java @@ -0,0 +1,143 @@ +/* + * 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.hudi.utilities.deser; + +import org.apache.hudi.DataSourceWriteOptions; +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.utilities.sources.helpers.SchemaTestProvider; +import org.apache.hudi.utilities.testutils.UtilitiesTestBase; + +import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig; +import io.confluent.kafka.serializers.KafkaAvroSerializer; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.generic.IndexedRecord; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Properties; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** + * Tests {@link KafkaAvroSchemaDeserializer}. + */ +public class TestKafkaAvroSchemaDeserializer extends UtilitiesTestBase { + + private final SchemaRegistryClient schemaRegistry; + private final KafkaAvroSerializer avroSerializer; + private final String topic; + private final Schema origSchema = createUserSchema(); + private final Schema evolSchema = createExtendUserSchema(); + private Properties defaultConfig = new Properties(); + + public TestKafkaAvroSchemaDeserializer() { + defaultConfig.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "bogus"); + defaultConfig.put("hoodie.deltastreamer.schemaprovider.class", SchemaTestProvider.class.getName()); + schemaRegistry = new MockSchemaRegistryClient(); + avroSerializer = new KafkaAvroSerializer(schemaRegistry, new HashMap(defaultConfig)); + topic = "test"; + } + + private Schema createUserSchema() { + String userSchema = "{\"namespace\": \"example.avro\", \"type\": \"record\", " + + "\"name\": \"User\"," + + "\"fields\": [{\"name\": \"name\", \"type\": \"string\"}]}"; + Schema.Parser parser = new Schema.Parser(); + Schema schema = parser.parse(userSchema); + return schema; + } + + private IndexedRecord createUserRecord() { + Schema schema = createUserSchema(); + GenericRecord avroRecord = new GenericData.Record(schema); + avroRecord.put("name", "testUser"); + return avroRecord; + } + + private Schema createExtendUserSchema() { + String userSchema = "{\"namespace\": \"example.avro\", \"type\": \"record\", " + + "\"name\": \"User\"," + + "\"fields\": [{\"name\": \"name\", \"type\": \"string\"}, " + + "{\"name\": \"age\", \"type\": [\"null\", \"int\"], \"default\": null}]}"; + Schema.Parser parser = new Schema.Parser(); + Schema schema = parser.parse(userSchema); + return schema; + } + + private IndexedRecord createExtendUserRecord() { + Schema schema = createExtendUserSchema(); + GenericRecord avroRecord = new GenericData.Record(schema); + avroRecord.put("name", "testUser"); + avroRecord.put("age", 30); + return avroRecord; + } + + /** + * Tests {@link KafkaAvroSchemaDeserializer#deserialize(boolean, String, Boolean, byte[], Schema)}. + */ + @Test + public void testKafkaAvroSchemaDeserializer() { + defaultConfig.put(DataSourceWriteOptions.JAVA_SPARK_CONTEXT_PROP(), jsc); + byte[] bytesOrigRecord; + IndexedRecord avroRecord = createUserRecord(); + SchemaTestProvider.schemaToReturn.set(origSchema); + KafkaAvroSchemaDeserializer avroDeserializer = new KafkaAvroSchemaDeserializer(schemaRegistry, new HashMap(defaultConfig)); + avroDeserializer.configure(new HashMap(defaultConfig), false); + bytesOrigRecord = avroSerializer.serialize(topic, avroRecord); + // record is serialized in orig schema and deserialized using same schema. + assertEquals(avroRecord, avroDeserializer.deserialize(false, topic, false, bytesOrigRecord, origSchema)); + + IndexedRecord avroRecordWithAllField = createExtendUserRecord(); + byte[] bytesExtendedRecord = avroSerializer.serialize(topic, avroRecordWithAllField); + + SchemaTestProvider.schemaToReturn.set(evolSchema); + avroDeserializer = new KafkaAvroSchemaDeserializer(schemaRegistry, new HashMap(defaultConfig)); + avroDeserializer.configure(new HashMap(defaultConfig), false); + // record is serialized w/ evolved schema, and deserialized w/ evolved schema + IndexedRecord avroRecordWithAllFieldActual = (IndexedRecord) avroDeserializer.deserialize(false, topic, false, bytesExtendedRecord, evolSchema); + assertEquals(avroRecordWithAllField, avroRecordWithAllFieldActual); + assertEquals(avroRecordWithAllFieldActual.getSchema(), evolSchema); + + // read old record w/ evolved schema. + IndexedRecord actualRec = (IndexedRecord) avroDeserializer.deserialize(false, topic, false, bytesOrigRecord, origSchema); + // record won't be equal to original record as we read w/ evolved schema. "age" will be added w/ default value of null + assertNotEquals(avroRecord, actualRec); + GenericRecord genericRecord = (GenericRecord) actualRec; + GenericRecord origGenRec = (GenericRecord) avroRecord; + assertEquals(genericRecord.get("name").toString(), origGenRec.get("name").toString()); + assertEquals(actualRec.getSchema(), evolSchema); + assertNull(genericRecord.get("age")); + } + + protected TypedProperties getConvertToTypedProperties(Map configs) { + TypedProperties typedProperties = new TypedProperties(); + for (Entry entry : configs.entrySet()) { + typedProperties.put(entry.getKey(), entry.getValue()); + } + return typedProperties; + } +} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java deleted file mode 100644 index 33e7d17b140b4..0000000000000 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java +++ /dev/null @@ -1,187 +0,0 @@ -/* - * 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.hudi.utilities.sources; - -import org.apache.hudi.common.config.TypedProperties; -import org.apache.hudi.common.testutils.HoodieTestDataGenerator; -import org.apache.hudi.common.util.Option; -import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamerMetrics; -import org.apache.hudi.utilities.deltastreamer.SourceFormatAdapter; -import org.apache.hudi.utilities.schema.FilebasedSchemaProvider; -import org.apache.hudi.utilities.sources.helpers.KafkaAvroTestDeserializer; -import org.apache.hudi.utilities.sources.helpers.KafkaAvroTestSerializer; -import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen.Config; -import org.apache.hudi.utilities.sources.helpers.SchemaTestProvider; -import org.apache.hudi.utilities.testutils.UtilitiesTestBase; - -import org.apache.avro.generic.GenericRecord; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; -import org.apache.spark.api.java.JavaRDD; -import org.apache.spark.streaming.kafka010.KafkaTestUtils; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.Mock; -import org.mockito.MockitoAnnotations; - -import java.util.List; -import java.util.UUID; -import java.util.stream.Collectors; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -/** - * Unit tests {@link AvroKakfaSource}. - */ -public class TestAvroKafkaSource extends UtilitiesTestBase { - - private static String TEST_TOPIC_NAME = "hoodie_test"; - - private FilebasedSchemaProvider schemaProvider; - private KafkaTestUtils testUtils; - private Producer kafkaProducer; - private HoodieDeltaStreamerMetrics metrics = mock(HoodieDeltaStreamerMetrics.class); - @Mock - private SchemaTestProvider schemaTestProvider; - - @BeforeAll - public static void initClass() throws Exception { - UtilitiesTestBase.initClass(); - } - - @AfterAll - public static void cleanupClass() { - UtilitiesTestBase.cleanupClass(); - } - - @BeforeEach - public void setup() throws Exception { - MockitoAnnotations.initMocks(this); - super.setup(); - schemaProvider = new FilebasedSchemaProvider(Helpers.setupSchemaOnDFS(), jsc); - testUtils = new KafkaTestUtils(); - testUtils.setup(); - kafkaProducer = createProducer(testUtils.brokerAddress(), "dummy"); - } - - @AfterEach - public void teardown() throws Exception { - super.teardown(); - testUtils.teardown(); - } - - private TypedProperties createPropsForAvroKafkaSource(Long maxEventsToReadFromKafkaSource, String resetStrategy) { - TypedProperties props = new TypedProperties(); - props.setProperty("hoodie.deltastreamer.source.kafka.topic", TEST_TOPIC_NAME); - props.setProperty("bootstrap.servers", testUtils.brokerAddress()); - props.setProperty("hoodie.deltastreamer.source.kafka.auto.reset.offsets", resetStrategy); - props.setProperty("hoodie.deltastreamer.kafka.source.maxEvents", - maxEventsToReadFromKafkaSource != null ? String.valueOf(maxEventsToReadFromKafkaSource) : - String.valueOf(Config.maxEventsFromKafkaSource)); - props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString()); - return props; - } - - public static Producer createProducer(String brokers, String schemaRegistryUrl) { - TypedProperties props = new TypedProperties(); - props.put("bootstrap.servers", brokers); - props.put("key.serializer", StringSerializer.class); - props.put("value.serializer", KafkaAvroTestSerializer.class); - props.put("schema.registry.url", schemaRegistryUrl); - return new KafkaProducer(props); - } - - @Test - public void testAvroKafkaSource() { - testUtils.createTopic(TEST_TOPIC_NAME, 2); - HoodieTestDataGenerator dataGenerator = new HoodieTestDataGenerator(); - TypedProperties props = createPropsForAvroKafkaSource(null, "earliest"); - props.put("value.deserializer", KafkaAvroTestDeserializer.class); - props.put("hoodie.deltastreamer.source.kafka.value.deserializer.class", KafkaAvroTestDeserializer.class.getCanonicalName()); - props.put("hoodie.deltastreamer.schemaprovider.class", SchemaTestProvider.class); - when(schemaTestProvider.getSourceSchema()).thenReturn(HoodieTestDataGenerator.AVRO_SCHEMA); - - Source avroKafkaSource = new AvroKafkaSource(props, jsc, sparkSession, schemaProvider, metrics); - SourceFormatAdapter kafkaSource = new SourceFormatAdapter(avroKafkaSource); - - // 1. Extract without any checkpoint => get all the data, respecting sourceLimit - assertEquals(Option.empty(), kafkaSource.fetchNewDataInAvroFormat(Option.empty(), Long.MAX_VALUE).getBatch()); - List genericRecordList = Helpers.toGenericRecords(dataGenerator.generateInserts("000", 10)); - for (GenericRecord genericRecord : genericRecordList) { - kafkaProducer.send(new ProducerRecord<>(TEST_TOPIC_NAME, genericRecord)); - } - kafkaProducer.flush(); - kafkaProducer.close(); - - InputBatch> fetch1 = kafkaSource.fetchNewDataInAvroFormat(Option.empty(), 10); - List actualGenRecs = fetch1.getBatch().get().collect(); - assertEquals(10, actualGenRecs.size()); - // to be fixed : equality checks for generic Records. - // assertEquals(genericRecordList, actualGenRecs); - assertListEquality(genericRecordList, actualGenRecs); - - // evolve schema and add more records to kafka - KafkaAvroTestSerializer.schemaToReturn = HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA; - KafkaAvroTestDeserializer.schemaToReturn = HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA; - when(schemaTestProvider.getSourceSchema()).thenReturn(HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA); - kafkaProducer = createProducer(testUtils.brokerAddress(), "dummy"); - - genericRecordList = Helpers.toGenericRecords(dataGenerator.generateInsertsStream("001", 20, false, - HoodieTestDataGenerator.TRIP_EVOLVED_EXAMPLE_SCHEMA).collect(Collectors.toList()), HoodieTestDataGenerator.AVRO_EVOLVED_SCHEMA); - for (GenericRecord genericRecord : genericRecordList) { - kafkaProducer.send(new ProducerRecord<>(TEST_TOPIC_NAME, genericRecord)); - } - kafkaProducer.flush(); - - InputBatch> fetch2 = kafkaSource.fetchNewDataInAvroFormat(Option.of(fetch1.getCheckpointForNextBatch()), 20); - List actualGenRecs2 = fetch2.getBatch().get().collect(); - assertEquals(20, actualGenRecs2.size()); - assertEquals(genericRecordList, actualGenRecs2); - } - - private void assertListEquality(List list1, List list2) { - System.out.println("Expected :: "); - for (GenericRecord gRec : list1) { - System.out.println("exp rec:: " + gRec.toString()); - } - System.out.println("Actual :: "); - for (GenericRecord gRec : list2) { - System.out.println("act rec:: " + gRec.toString()); - } - assertEquals(list1.size(), list2.size()); - for (GenericRecord genericRecord : list1) { - System.out.println("Checking " + genericRecord.toString() + " from exp "); - assertTrue(list2.contains(genericRecord)); - } - for (GenericRecord genericRecord : list2) { - System.out.println("Checking " + genericRecord.toString() + " from actual "); - assertTrue(list1.contains(genericRecord)); - } - } - -} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java deleted file mode 100644 index dafe92aeec375..0000000000000 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestCustomDeserializer.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * 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.hudi.utilities.sources.helpers; - -import org.apache.hudi.common.config.TypedProperties; -import org.apache.hudi.common.testutils.HoodieTestDataGenerator; -import org.apache.hudi.common.util.ReflectionUtils; -import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.utilities.deser.KafkaAvroSchemaDeserializer; -import org.apache.hudi.utilities.schema.SchemaProvider; - -import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; -import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; -import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; -import org.apache.avro.Schema; - -import java.io.IOException; -import java.util.Map; -import java.util.Objects; - -/** - * KafkaAvroSchemaDeserializer for tests using {@link MockSchemaRegistryClient}. Since in tests we can't use schema registry, - * have to override calls especially super.configure(). So, had to introduce these test classes. Impl of configure() should - * be in line with {@link KafkaAvroSchemaDeserializer} except for super.configure(). - */ -public class KafkaAvroTestCustomDeserializer extends KafkaAvroSchemaDeserializer { - - public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; - - public KafkaAvroTestCustomDeserializer() { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - } - - public KafkaAvroTestCustomDeserializer(SchemaRegistryClient client) { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - } - - public KafkaAvroTestCustomDeserializer(SchemaRegistryClient client, Map props) { - this.schemaRegistry = client; - this.configure(this.deserializerConfig(props)); - } - - public void configure(Map configs, boolean isKey) { - try { - TypedProperties props = getConvertToTypedProperties(configs); - String className = props.getString("hoodie.deltastreamer.schemaprovider.class"); - SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props); - sourceSchema = Objects.requireNonNull(schemaProvider).getSourceSchema(); - } catch (Throwable e) { - throw new HoodieException(e); - } - } -} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java deleted file mode 100644 index 15fbc66042c7e..0000000000000 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestDeserializer.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * 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.hudi.utilities.sources.helpers; - -import org.apache.hudi.common.testutils.HoodieTestDataGenerator; - -import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; -import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; -import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; -import io.confluent.kafka.serializers.KafkaAvroDeserializer; -import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig; -import org.apache.avro.Schema; - -import java.io.IOException; -import java.util.Map; - -/** - * KafkaAvroDeserializer for tests using {@link MockSchemaRegistryClient}. Since in tests we can't use schema registry, have to override calls especially - * super.configure(). So, had to introduce these test classes. Impl of configure() should be in line with - * {@link KafkaAvroDeserializer} except for super.configure(). - */ -public class KafkaAvroTestDeserializer extends KafkaAvroDeserializer { - - public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; - - public KafkaAvroTestDeserializer() { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - } - - public KafkaAvroTestDeserializer(SchemaRegistryClient client) { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - } - - public KafkaAvroTestDeserializer(SchemaRegistryClient client, Map props) { - this.schemaRegistry = client; - this.configure(this.deserializerConfig(props)); - } - - protected void configure(KafkaAvroDeserializerConfig config) { - // no op - } - - public void configure(Map configs, boolean isKey) { - } -} \ No newline at end of file diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java deleted file mode 100644 index 59a628b5f0097..0000000000000 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/KafkaAvroTestSerializer.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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.hudi.utilities.sources.helpers; - -import org.apache.hudi.common.testutils.HoodieTestDataGenerator; - -import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; -import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; -import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; -import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig; -import io.confluent.kafka.serializers.KafkaAvroSerializer; -import org.apache.avro.Schema; - -import java.io.IOException; -import java.util.HashMap; -import java.util.Map; - -/** - * KafkaAvroSerializer for tests using {@link MockSchemaRegistryClient}. Since in tests we can't use schema registry, - * have to override calls especially super.configure(). So, had to introduce these test classes. Impl of configure() - * should be in line with {@link KafkaAvroSerializer} except for super.configure(). - */ -public class KafkaAvroTestSerializer extends KafkaAvroSerializer { - - public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; - - public KafkaAvroTestSerializer() { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - } - - public KafkaAvroTestSerializer(SchemaRegistryClient client) { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - } - - public KafkaAvroTestSerializer(SchemaRegistryClient client, Map props) { - this.schemaRegistry = new MockSchemaRegistryClient() { - @Override - public synchronized Schema getByID(int id) throws IOException, RestClientException { - return schemaToReturn; - } - }; - Map map = new HashMap<>(); - map.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "dummy"); - this.configure(this.serializerConfig(props)); - } - -} diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java index b9ee464da0433..b0016130116e4 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java @@ -25,19 +25,22 @@ import org.apache.avro.Schema; import org.apache.spark.api.java.JavaSparkContext; +import java.util.concurrent.atomic.AtomicReference; + /** * {@link SchemaProvider} for tests. */ public class SchemaTestProvider extends SchemaProvider { - public static Schema schemaToReturn = HoodieTestDataGenerator.AVRO_SCHEMA; + public static AtomicReference schemaToReturn = new AtomicReference<>(HoodieTestDataGenerator.AVRO_SCHEMA); - protected SchemaTestProvider(TypedProperties props, JavaSparkContext jssc) { - super(props, jssc); + public SchemaTestProvider(TypedProperties props, JavaSparkContext jsc) { + super(props, jsc); } @Override public Schema getSourceSchema() { - return schemaToReturn; + return schemaToReturn.get(); } + } From b73c0e31ed6dc483bbfcafdcf41ab740dd271ef4 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Fri, 19 Mar 2021 18:42:34 -0400 Subject: [PATCH 8/8] Addressing feedback --- .../main/scala/org/apache/hudi/DataSourceOptions.scala | 3 +-- .../utilities/deser/KafkaAvroSchemaDeserializer.java | 4 +--- .../apache/hudi/utilities/schema/SchemaProvider.java | 4 ++++ .../apache/hudi/utilities/sources/AvroKafkaSource.java | 10 ++++++---- .../deser/TestKafkaAvroSchemaDeserializer.java | 2 -- .../utilities/sources/helpers/SchemaTestProvider.java | 5 ++--- 6 files changed, 14 insertions(+), 14 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala index 42e303c58b266..51f32a2a3847a 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala @@ -376,9 +376,8 @@ object DataSourceWriteOptions { // Avro Kafka Source configs val KAFKA_AVRO_VALUE_DESERIALIZER = "hoodie.deltastreamer.source.kafka.value.deserializer.class" - // val DEFAULT_KAFKA_AVRO_VALUE_DESERIALIZER = classOf[io.confluent.kafka.serializers.KafkaAvroDeserializer] + // Schema provider class to be set to be used in custom kakfa deserializer val SCHEMA_PROVIDER_CLASS_PROP = "hoodie.deltastreamer.schemaprovider.class" - val JAVA_SPARK_CONTEXT_PROP = "java.spark.context" } diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java index 9d429de834409..5d0a116cab945 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -28,7 +28,6 @@ import org.apache.hudi.exception.HoodieException; import org.apache.hudi.utilities.schema.SchemaProvider; import org.apache.kafka.common.errors.SerializationException; -import org.apache.spark.api.java.JavaSparkContext; import java.util.Map; import java.util.Map.Entry; @@ -53,8 +52,7 @@ public void configure(Map configs, boolean isKey) { try { TypedProperties props = getConvertToTypedProperties(configs); String className = props.getString(DataSourceWriteOptions.SCHEMA_PROVIDER_CLASS_PROP()); - JavaSparkContext jsc = (JavaSparkContext) props.get(DataSourceWriteOptions.JAVA_SPARK_CONTEXT_PROP()); - SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props, jsc); + SchemaProvider schemaProvider = (SchemaProvider) ReflectionUtils.loadClass(className, props); sourceSchema = Objects.requireNonNull(schemaProvider).getSourceSchema(); } catch (Throwable e) { throw new HoodieException(e); diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaProvider.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaProvider.java index c653622b9a0c7..bcbdbf049be3f 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaProvider.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaProvider.java @@ -38,6 +38,10 @@ public abstract class SchemaProvider implements Serializable { protected JavaSparkContext jssc; + public SchemaProvider(TypedProperties props) { + this(props, null); + } + protected SchemaProvider(TypedProperties props, JavaSparkContext jssc) { this.config = props; this.jssc = jssc; diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java index bc7c42cf1e438..511a72c280e8a 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/AvroKafkaSource.java @@ -46,6 +46,9 @@ public class AvroKafkaSource extends AvroSource { private static final Logger LOG = LogManager.getLogger(AvroKafkaSource.class); + // these are native kafka's config. do not change the config names. + private static final String NATIVE_KAFKA_KEY_DESERIALIZER_PROP = "key.deserializer"; + private static final String NATIVE_KAFKA_VALUE_DESERIALIZER_PROP = "value.deserializer"; private final KafkaOffsetGen offsetGen; private final HoodieDeltaStreamerMetrics metrics; @@ -53,19 +56,18 @@ public AvroKafkaSource(TypedProperties props, JavaSparkContext sparkContext, Spa SchemaProvider schemaProvider, HoodieDeltaStreamerMetrics metrics) { super(props, sparkContext, sparkSession, schemaProvider); - props.put("key.deserializer", StringDeserializer.class); + props.put(NATIVE_KAFKA_KEY_DESERIALIZER_PROP, StringDeserializer.class); String deserializerClassName = props.getString(DataSourceWriteOptions.KAFKA_AVRO_VALUE_DESERIALIZER(), ""); if (deserializerClassName.isEmpty()) { - props.put("value.deserializer", KafkaAvroDeserializer.class); + props.put(NATIVE_KAFKA_VALUE_DESERIALIZER_PROP, KafkaAvroDeserializer.class); } else { try { if (schemaProvider == null) { throw new HoodieIOException("SchemaProvider has to be set to use custom Deserializer"); } props.put(DataSourceWriteOptions.SCHEMA_PROVIDER_CLASS_PROP(), schemaProvider.getClass().getName()); - props.put(DataSourceWriteOptions.JAVA_SPARK_CONTEXT_PROP(), sparkContext); - props.put("value.deserializer", Class.forName(deserializerClassName)); + props.put(NATIVE_KAFKA_VALUE_DESERIALIZER_PROP, Class.forName(deserializerClassName)); } catch (ClassNotFoundException e) { String error = "Could not load custom avro kafka deserializer: " + deserializerClassName; LOG.error(error); diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java index beee0d3c3c49e..14c5f01079d88 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java @@ -18,7 +18,6 @@ package org.apache.hudi.utilities.deser; -import org.apache.hudi.DataSourceWriteOptions; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.utilities.sources.helpers.SchemaTestProvider; import org.apache.hudi.utilities.testutils.UtilitiesTestBase; @@ -101,7 +100,6 @@ private IndexedRecord createExtendUserRecord() { */ @Test public void testKafkaAvroSchemaDeserializer() { - defaultConfig.put(DataSourceWriteOptions.JAVA_SPARK_CONTEXT_PROP(), jsc); byte[] bytesOrigRecord; IndexedRecord avroRecord = createUserRecord(); SchemaTestProvider.schemaToReturn.set(origSchema); diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java index b0016130116e4..5f3c7ed528918 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/SchemaTestProvider.java @@ -23,7 +23,6 @@ import org.apache.hudi.utilities.schema.SchemaProvider; import org.apache.avro.Schema; -import org.apache.spark.api.java.JavaSparkContext; import java.util.concurrent.atomic.AtomicReference; @@ -34,8 +33,8 @@ public class SchemaTestProvider extends SchemaProvider { public static AtomicReference schemaToReturn = new AtomicReference<>(HoodieTestDataGenerator.AVRO_SCHEMA); - public SchemaTestProvider(TypedProperties props, JavaSparkContext jsc) { - super(props, jsc); + public SchemaTestProvider(TypedProperties props) { + super(props); } @Override