This is an automated email from the ASF dual-hosted git repository.
codope pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 9cfbbf25dfb [HUDI-7416] Add interface for SourceProfile to be used in
StreamSync for reading data (#10736)
9cfbbf25dfb is described below
commit 9cfbbf25dfb84ab5294f420b2b7535a76afdb068
Author: Vinish Reddy <[email protected]>
AuthorDate: Mon Feb 26 20:44:50 2024 +0530
[HUDI-7416] Add interface for SourceProfile to be used in StreamSync for
reading data (#10736)
Introducing a new class known as SourceProfile which contains details about
how the next sync round in StreamSync should be consumed. For eg:
KafkaSourceProfile contains number of events to consume in this sync round.
S3SourceProfile contains the list of files to consume in this sync round
HudiIncrementalSourceProfile contains the beginInstant and endInstant
commit times to consume in this sync round.
In future we can add the method for choosing the writeOperationType and
indexType as well, for sourceProfile.getSourceSpecificContext() will be used to
consume the data from the source.
---
.../org/apache/hudi/utilities/UtilHelpers.java | 19 ++++++++
.../hudi/utilities/deltastreamer/DeltaSync.java | 4 +-
.../hudi/utilities/sources/AvroKafkaSource.java | 13 ++++--
.../hudi/utilities/sources/JsonKafkaSource.java | 14 ++++--
.../apache/hudi/utilities/sources/KafkaSource.java | 39 +++++++++++-----
.../hudi/utilities/sources/ProtoKafkaSource.java | 15 ++++--
.../org/apache/hudi/utilities/sources/Source.java | 11 ++++-
.../utilities/sources/helpers/KafkaOffsetGen.java | 39 ++++++++--------
.../utilities/streamer/DefaultStreamContext.java | 48 +++++++++++++++++++
.../hudi/utilities/streamer/HoodieStreamer.java | 21 ++++++---
.../hudi/utilities/streamer/SourceProfile.java | 54 ++++++++++++++++++++++
.../utilities/streamer/SourceProfileSupplier.java | 34 ++++++++++++++
.../hudi/utilities/streamer/StreamContext.java | 44 ++++++++++++++++++
.../apache/hudi/utilities/streamer/StreamSync.java | 10 ++--
.../utilities/sources/BaseTestKafkaSource.java | 51 ++++++++++++++++++++
.../utilities/sources/TestJsonKafkaSource.java | 17 +++++--
.../utilities/sources/TestProtoKafkaSource.java | 3 +-
17 files changed, 375 insertions(+), 61 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
index 18e92a8463c..650634972d5 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java
@@ -66,6 +66,7 @@ import org.apache.hudi.utilities.sources.InputBatch;
import org.apache.hudi.utilities.sources.Source;
import
org.apache.hudi.utilities.sources.processor.ChainedJsonKafkaSourcePostProcessor;
import
org.apache.hudi.utilities.sources.processor.JsonKafkaSourcePostProcessor;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.hudi.utilities.transform.ChainedTransformer;
import org.apache.hudi.utilities.transform.ErrorTableAwareChainedTransformer;
import org.apache.hudi.utilities.transform.Transformer;
@@ -156,6 +157,24 @@ public class UtilHelpers {
}
}
+ public static Source createSource(String sourceClass, TypedProperties cfg,
JavaSparkContext jssc,
+ SparkSession sparkSession,
HoodieIngestionMetrics metrics, StreamContext streamContext)
+ throws IOException {
+ try {
+ try {
+ return (Source) ReflectionUtils.loadClass(sourceClass,
+ new Class<?>[] {TypedProperties.class, JavaSparkContext.class,
+ SparkSession.class,
+ HoodieIngestionMetrics.class, StreamContext.class},
+ cfg, jssc, sparkSession, metrics, streamContext);
+ } catch (HoodieException e) {
+ return createSource(sourceClass, cfg, jssc, sparkSession,
streamContext.getSchemaProvider(), metrics);
+ }
+ } catch (Throwable e) {
+ throw new IOException("Could not load source class " + sourceClass, e);
+ }
+ }
+
public static JsonKafkaSourcePostProcessor
createJsonKafkaSourcePostProcessor(String postProcessorClassNames,
TypedProperties props) throws IOException {
if (StringUtils.isNullOrEmpty(postProcessorClassNames)) {
return null;
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index c794db32510..4002d1579bb 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -22,7 +22,9 @@ package org.apache.hudi.utilities.deltastreamer;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.HoodieStreamer;
import org.apache.hudi.utilities.streamer.StreamSync;
@@ -49,6 +51,6 @@ public class DeltaSync extends StreamSync {
public DeltaSync(HoodieDeltaStreamer.Config cfg, SparkSession sparkSession,
SchemaProvider schemaProvider,
TypedProperties props, HoodieSparkEngineContext
hoodieSparkContext, FileSystem fs, Configuration conf,
Function<SparkRDDWriteClient, Boolean>
onInitializingHoodieWriteClient) throws IOException {
- super(cfg, sparkSession, schemaProvider, props, hoodieSparkContext, fs,
conf, onInitializingHoodieWriteClient);
+ super(cfg, sparkSession, props, hoodieSparkContext, fs, conf,
onInitializingHoodieWriteClient, new DefaultStreamContext(schemaProvider,
Option.empty()));
}
}
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 2bf92280faf..36c83d63030 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
@@ -27,6 +27,8 @@ import
org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.AvroConvertor;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -69,10 +71,13 @@ public class AvroKafkaSource extends
KafkaSource<GenericRecord> {
public AvroKafkaSource(TypedProperties props, JavaSparkContext sparkContext,
SparkSession sparkSession,
SchemaProvider schemaProvider, HoodieIngestionMetrics
metrics) {
- super(props, sparkContext, sparkSession,
- UtilHelpers.getSchemaProviderForKafkaSource(schemaProvider, props,
sparkContext),
- SourceType.AVRO, metrics);
- this.originalSchemaProvider = schemaProvider;
+ this(props, sparkContext, sparkSession, metrics, new
DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ public AvroKafkaSource(TypedProperties properties, JavaSparkContext
sparkContext, SparkSession sparkSession, HoodieIngestionMetrics metrics,
StreamContext streamContext) {
+ super(properties, sparkContext, sparkSession, SourceType.AVRO, metrics,
+ new
DefaultStreamContext(UtilHelpers.getSchemaProviderForKafkaSource(streamContext.getSchemaProvider(),
properties, sparkContext), streamContext.getSourceProfileSupplier()));
+ this.originalSchemaProvider = streamContext.getSchemaProvider();
props.put(NATIVE_KAFKA_KEY_DESERIALIZER_PROP,
StringDeserializer.class.getName());
deserializerClassName = getStringWithAltKeys(props,
KAFKA_AVRO_VALUE_DESERIALIZER_CLASS, true);
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
index eb67abfee3a..6e95a315260 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKafkaSource.java
@@ -19,6 +19,7 @@
package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.JsonKafkaPostProcessorConfig;
@@ -27,6 +28,8 @@ import
org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
import
org.apache.hudi.utilities.sources.processor.JsonKafkaSourcePostProcessor;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.StreamContext;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -44,10 +47,10 @@ import java.util.LinkedList;
import java.util.List;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
+import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_OFFSET_COLUMN;
import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_PARTITION_COLUMN;
import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_TIMESTAMP_COLUMN;
-import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
/**
* Read json kafka data.
@@ -56,9 +59,12 @@ public class JsonKafkaSource extends KafkaSource<String> {
public JsonKafkaSource(TypedProperties properties, JavaSparkContext
sparkContext, SparkSession sparkSession,
SchemaProvider schemaProvider, HoodieIngestionMetrics
metrics) {
- super(properties, sparkContext, sparkSession,
- UtilHelpers.getSchemaProviderForKafkaSource(schemaProvider,
properties, sparkContext),
- SourceType.JSON, metrics);
+ this(properties, sparkContext, sparkSession, metrics, new
DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ public JsonKafkaSource(TypedProperties properties, JavaSparkContext
sparkContext, SparkSession sparkSession, HoodieIngestionMetrics metrics,
StreamContext streamContext) {
+ super(properties, sparkContext, sparkSession, SourceType.JSON, metrics,
+ new
DefaultStreamContext(UtilHelpers.getSchemaProviderForKafkaSource(streamContext.getSchemaProvider(),
properties, sparkContext), streamContext.getSourceProfileSupplier()));
properties.put("key.deserializer", StringDeserializer.class.getName());
properties.put("value.deserializer", StringDeserializer.class.getName());
this.offsetGen = new KafkaOffsetGen(props);
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
index bb26d579582..52a6a1217cc 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/KafkaSource.java
@@ -26,6 +26,8 @@ import
org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
+import org.apache.hudi.utilities.streamer.SourceProfile;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
@@ -50,9 +52,9 @@ abstract class KafkaSource<T> extends Source<JavaRDD<T>> {
protected final boolean shouldAddOffsets;
protected KafkaSource(TypedProperties props, JavaSparkContext sparkContext,
SparkSession sparkSession,
- SchemaProvider schemaProvider, SourceType sourceType,
HoodieIngestionMetrics metrics) {
- super(props, sparkContext, sparkSession, schemaProvider, sourceType);
- this.schemaProvider = schemaProvider;
+ SourceType sourceType, HoodieIngestionMetrics metrics,
StreamContext streamContext) {
+ super(props, sparkContext, sparkSession, sourceType, streamContext);
+ this.schemaProvider = streamContext.getSchemaProvider();
this.metrics = metrics;
this.shouldAddOffsets =
KafkaOffsetPostProcessor.Config.shouldAddOffsets(props);
}
@@ -60,21 +62,34 @@ abstract class KafkaSource<T> extends Source<JavaRDD<T>> {
@Override
protected InputBatch<JavaRDD<T>> fetchNewData(Option<String>
lastCheckpointStr, long sourceLimit) {
try {
- OffsetRange[] offsetRanges =
offsetGen.getNextOffsetRanges(lastCheckpointStr, sourceLimit, metrics);
- long totalNewMsgs =
KafkaOffsetGen.CheckpointUtils.totalNewMessages(offsetRanges);
- LOG.info("About to read " + totalNewMsgs + " from Kafka for topic :" +
offsetGen.getTopicName());
- if (totalNewMsgs <= 0) {
-
metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT,
0);
- return new InputBatch<>(Option.empty(),
KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ OffsetRange[] offsetRanges;
+ if (sourceProfileSupplier.isPresent() &&
sourceProfileSupplier.get().getSourceProfile() != null) {
+ SourceProfile<Long> kafkaSourceProfile =
sourceProfileSupplier.get().getSourceProfile();
+ offsetRanges = offsetGen.getNextOffsetRanges(lastCheckpointStr,
kafkaSourceProfile.getSourceSpecificContext(),
kafkaSourceProfile.getSourcePartitions(), metrics);
+ LOG.info("About to read numEvents {} of size {} bytes in {} partitions
from Kafka for topic {} with offsetRanges {}",
+ kafkaSourceProfile.getSourceSpecificContext(),
kafkaSourceProfile.getMaxSourceBytes(),
+ kafkaSourceProfile.getSourcePartitions(),
offsetGen.getTopicName(), offsetRanges);
+ } else {
+ offsetRanges = offsetGen.getNextOffsetRanges(lastCheckpointStr,
sourceLimit, metrics);
}
-
metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT,
totalNewMsgs);
- JavaRDD<T> newDataRDD = toRDD(offsetRanges);
- return new InputBatch<>(Option.of(newDataRDD),
KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ return toInputBatch(offsetRanges);
} catch (org.apache.kafka.common.errors.TimeoutException e) {
throw new HoodieSourceTimeoutException("Kafka Source timed out " +
e.getMessage());
}
}
+ private InputBatch<JavaRDD<T>> toInputBatch(OffsetRange[] offsetRanges) {
+ long totalNewMsgs =
KafkaOffsetGen.CheckpointUtils.totalNewMessages(offsetRanges);
+ LOG.info("About to read " + totalNewMsgs + " from Kafka for topic :" +
offsetGen.getTopicName());
+ if (totalNewMsgs <= 0) {
+
metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT,
0);
+ return new InputBatch<>(Option.empty(),
KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ }
+
metrics.updateStreamerSourceNewMessageCount(METRIC_NAME_KAFKA_MESSAGE_IN_COUNT,
totalNewMsgs);
+ JavaRDD<T> newDataRDD = toRDD(offsetRanges);
+ return new InputBatch<>(Option.of(newDataRDD),
KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
+ }
+
abstract JavaRDD<T> toRDD(OffsetRange[] offsetRanges);
@Override
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
index 67927480454..d7a15b3932c 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/ProtoKafkaSource.java
@@ -19,12 +19,16 @@
package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.ProtoClassBasedSchemaProviderConfig;
import org.apache.hudi.utilities.exception.HoodieReadFromSourceException;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KafkaOffsetGen;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.StreamContext;
import com.google.protobuf.Message;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -51,9 +55,14 @@ public class ProtoKafkaSource extends KafkaSource<Message> {
private final String className;
- public ProtoKafkaSource(TypedProperties props, JavaSparkContext sparkContext,
- SparkSession sparkSession, SchemaProvider
schemaProvider, HoodieIngestionMetrics metrics) {
- super(props, sparkContext, sparkSession, schemaProvider, SourceType.PROTO,
metrics);
+ public ProtoKafkaSource(TypedProperties props, JavaSparkContext
sparkContext, SparkSession sparkSession,
+ SchemaProvider schemaProvider,
HoodieIngestionMetrics metrics) {
+ this(props, sparkContext, sparkSession, metrics, new
DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ public ProtoKafkaSource(TypedProperties properties, JavaSparkContext
sparkContext, SparkSession sparkSession, HoodieIngestionMetrics metrics,
StreamContext streamContext) {
+ super(properties, sparkContext, sparkSession, SourceType.PROTO, metrics,
+ new
DefaultStreamContext(UtilHelpers.getSchemaProviderForKafkaSource(streamContext.getSchemaProvider(),
properties, sparkContext), streamContext.getSourceProfileSupplier()));
checkRequiredConfigProperties(props, Collections.singletonList(
ProtoClassBasedSchemaProviderConfig.PROTO_SCHEMA_CLASS_NAME));
props.put(NATIVE_KAFKA_KEY_DESERIALIZER_PROP, StringDeserializer.class);
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
index cbc0722056b..dfb07c718a0 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
@@ -25,6 +25,9 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.utilities.callback.SourceCommitCallback;
import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
+import org.apache.hudi.utilities.streamer.SourceProfileSupplier;
+import org.apache.hudi.utilities.streamer.StreamContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
@@ -44,6 +47,7 @@ public abstract class Source<T> implements
SourceCommitCallback, Serializable {
protected transient TypedProperties props;
protected transient JavaSparkContext sparkContext;
protected transient SparkSession sparkSession;
+ protected transient Option<SourceProfileSupplier> sourceProfileSupplier;
private transient SchemaProvider overriddenSchemaProvider;
private final SourceType sourceType;
@@ -55,11 +59,16 @@ public abstract class Source<T> implements
SourceCommitCallback, Serializable {
protected Source(TypedProperties props, JavaSparkContext sparkContext,
SparkSession sparkSession,
SchemaProvider schemaProvider, SourceType sourceType) {
+ this(props, sparkContext, sparkSession, sourceType, new
DefaultStreamContext(schemaProvider, Option.empty()));
+ }
+
+ protected Source(TypedProperties props, JavaSparkContext sparkContext,
SparkSession sparkSession, SourceType sourceType, StreamContext streamContext) {
this.props = props;
this.sparkContext = sparkContext;
this.sparkSession = sparkSession;
- this.overriddenSchemaProvider = schemaProvider;
+ this.overriddenSchemaProvider = streamContext.getSchemaProvider();
this.sourceType = sourceType;
+ this.sourceProfileSupplier = streamContext.getSourceProfileSupplier();
}
@PublicAPIMethod(maturity = ApiMaturityLevel.STABLE)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
index d5faec3595e..32df651d556 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java
@@ -241,7 +241,24 @@ public class KafkaOffsetGen {
}
public OffsetRange[] getNextOffsetRanges(Option<String> lastCheckpointStr,
long sourceLimit, HoodieIngestionMetrics metrics) {
+ // Come up with final set of OffsetRanges to read (account for new
partitions, limit number of events)
+ long maxEventsToReadFromKafka = getLongWithAltKeys(props,
KafkaSourceConfig.MAX_EVENTS_FROM_KAFKA_SOURCE);
+ long numEvents;
+ if (sourceLimit == Long.MAX_VALUE) {
+ numEvents = maxEventsToReadFromKafka;
+ LOG.info("SourceLimit not configured, set numEvents to default value : "
+ maxEventsToReadFromKafka);
+ } else {
+ numEvents = sourceLimit;
+ }
+
+ long minPartitions = getLongWithAltKeys(props,
KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS);
+ LOG.info("getNextOffsetRanges set config " +
KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS.key() + " to " + minPartitions);
+
+ return getNextOffsetRanges(lastCheckpointStr, numEvents, minPartitions,
metrics);
+ }
+
+ public OffsetRange[] getNextOffsetRanges(Option<String> lastCheckpointStr,
long numEvents, long minPartitions, HoodieIngestionMetrics metrics) {
// Obtain current metadata for the topic
Map<TopicPartition, Long> fromOffsets;
Map<TopicPartition, Long> toOffsets;
@@ -279,29 +296,9 @@ public class KafkaOffsetGen {
// Obtain the latest offsets.
toOffsets = consumer.endOffsets(topicPartitions);
}
-
- // Come up with final set of OffsetRanges to read (account for new
partitions, limit number of events)
- long maxEventsToReadFromKafka = getLongWithAltKeys(props,
KafkaSourceConfig.MAX_EVENTS_FROM_KAFKA_SOURCE);
-
- long numEvents;
- if (sourceLimit == Long.MAX_VALUE) {
- numEvents = maxEventsToReadFromKafka;
- LOG.info("SourceLimit not configured, set numEvents to default value : "
+ maxEventsToReadFromKafka);
- } else {
- numEvents = sourceLimit;
- }
-
- // TODO(HUDI-4625) remove
- if (numEvents < toOffsets.size()) {
- throw new HoodieException("sourceLimit should not be less than the
number of kafka partitions");
- }
-
- long minPartitions = getLongWithAltKeys(props,
KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS);
- LOG.info("getNextOffsetRanges set config " +
KafkaSourceConfig.KAFKA_SOURCE_MIN_PARTITIONS.key() + " to " + minPartitions);
-
return CheckpointUtils.computeOffsetRanges(fromOffsets, toOffsets,
numEvents, minPartitions);
}
-
+
/**
* Fetch partition infos for given topic.
*
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java
new file mode 100644
index 00000000000..f8dabeb89c9
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java
@@ -0,0 +1,48 @@
+/*
+ * 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.streamer;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+
+/**
+ * The default implementation for the StreamContext interface,
+ * composes SchemaProvider and SourceProfileSupplier currently,
+ * can be extended for other arguments in the future.
+ */
+public class DefaultStreamContext implements StreamContext {
+
+ private final SchemaProvider schemaProvider;
+ private final Option<SourceProfileSupplier> sourceProfileSupplier;
+
+ public DefaultStreamContext(SchemaProvider schemaProvider,
Option<SourceProfileSupplier> sourceProfileSupplier) {
+ this.schemaProvider = schemaProvider;
+ this.sourceProfileSupplier = sourceProfileSupplier;
+ }
+
+ @Override
+ public SchemaProvider getSchemaProvider() {
+ return schemaProvider;
+ }
+
+ @Override
+ public Option<SourceProfileSupplier> getSourceProfileSupplier() {
+ return sourceProfileSupplier;
+ }
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
index 8ecc937c5e7..ef31cc34ab5 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
@@ -143,8 +143,12 @@ public class HoodieStreamer implements Serializable {
this(cfg, jssc, fs, conf, Option.empty());
}
+ public HoodieStreamer(Config cfg, JavaSparkContext jssc, FileSystem fs,
Configuration conf, Option<TypedProperties> propsOverride) throws IOException {
+ this(cfg, jssc, fs, conf, propsOverride, Option.empty());
+ }
+
public HoodieStreamer(Config cfg, JavaSparkContext jssc, FileSystem fs,
Configuration conf,
- Option<TypedProperties> propsOverride) throws
IOException {
+ Option<TypedProperties> propsOverride,
Option<SourceProfileSupplier> sourceProfileSupplier) throws IOException {
this.properties = combineProperties(cfg, propsOverride,
jssc.hadoopConfiguration());
if (cfg.initialCheckpointProvider != null && cfg.checkpoint == null) {
InitialCheckPointProvider checkPointProvider =
@@ -158,7 +162,7 @@ public class HoodieStreamer implements Serializable {
cfg.runBootstrap ? new BootstrapExecutor(cfg, jssc, fs, conf,
this.properties) : null);
HoodieSparkEngineContext sparkEngineContext = new
HoodieSparkEngineContext(jssc);
this.ingestionService = Option.ofNullable(
- cfg.runBootstrap ? null : new StreamSyncService(cfg,
sparkEngineContext, fs, conf, Option.ofNullable(this.properties)));
+ cfg.runBootstrap ? null : new StreamSyncService(cfg,
sparkEngineContext, fs, conf, Option.ofNullable(this.properties),
sourceProfileSupplier));
}
private static TypedProperties combineProperties(Config cfg,
Option<TypedProperties> propsOverride, Configuration hadoopConf) {
@@ -656,7 +660,7 @@ public class HoodieStreamer implements Serializable {
private final Option<ConfigurationHotUpdateStrategy>
configurationHotUpdateStrategyOpt;
public StreamSyncService(Config cfg, HoodieSparkEngineContext
hoodieSparkContext, FileSystem fs, Configuration conf,
- Option<TypedProperties> properties) throws
IOException {
+ Option<TypedProperties> properties,
Option<SourceProfileSupplier> sourceProfileSupplier) throws IOException {
super(HoodieIngestionConfig.newBuilder()
.isContinuous(cfg.continuousMode)
.withMinSyncInternalSeconds(cfg.minSyncIntervalSeconds).build());
@@ -708,13 +712,18 @@ public class HoodieStreamer implements Serializable {
UtilHelpers.createSchemaProvider(cfg.schemaProviderClassName, props,
hoodieSparkContext.jsc()),
props, hoodieSparkContext.jsc(), cfg.transformerClassNames);
- streamSync = new StreamSync(cfg, sparkSession, schemaProvider, props,
hoodieSparkContext, fs, conf, this::onInitializingWriteClient);
+ streamSync = new StreamSync(cfg, sparkSession, props,
hoodieSparkContext, fs, conf, this::onInitializingWriteClient, new
DefaultStreamContext(schemaProvider, sourceProfileSupplier));
}
public StreamSyncService(HoodieStreamer.Config cfg,
HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf)
throws IOException {
- this(cfg, hoodieSparkContext, fs, conf, Option.empty());
+ this(cfg, hoodieSparkContext, fs, conf, Option.empty(), Option.empty());
+ }
+
+ public StreamSyncService(HoodieStreamer.Config cfg,
HoodieSparkEngineContext hoodieSparkContext, FileSystem fs, Configuration conf,
Option<TypedProperties> properties)
+ throws IOException {
+ this(cfg, hoodieSparkContext, fs, conf, properties, Option.empty());
}
private void initializeTableTypeAndBaseFileFormat() {
@@ -728,7 +737,7 @@ public class HoodieStreamer implements Serializable {
if (streamSync != null) {
streamSync.close();
}
- streamSync = new StreamSync(cfg, sparkSession, schemaProvider, props,
hoodieSparkContext, fs, hiveConf, this::onInitializingWriteClient);
+ streamSync = new StreamSync(cfg, sparkSession, props,
hoodieSparkContext, fs, hiveConf, this::onInitializingWriteClient, new
DefaultStreamContext(schemaProvider, Option.empty()));
}
@Override
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java
new file mode 100644
index 00000000000..d830cf5dee3
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfile.java
@@ -0,0 +1,54 @@
+/*
+ * 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.streamer;
+
+import org.apache.hudi.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIClass;
+import org.apache.hudi.PublicAPIMethod;
+
+/**
+ * A profile containing details about how the next input batch in StreamSync
should be consumed and written.
+ * For eg: KafkaSourceProfile contains number of events to consume in this
sync round.
+ * S3SourceProfile contains the list of files to consume in this sync round.
+ * HudiIncrementalSourceProfile contains the beginInstant and endInstant
commit times to consume in this sync round etc.
+ *
+ * @param <T> The type for source context, varies based on sourceType as
described above.
+ */
+@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
+public interface SourceProfile<T> {
+
+ /**
+ * @return The maxBytes that will be consumed from the source in this sync
round.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ long getMaxSourceBytes();
+
+ /**
+ * @return The number of output partitions required in source RDD.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ int getSourcePartitions();
+
+ /**
+ * @return The source specific context based on sourceType as described
above.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ T getSourceSpecificContext();
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java
new file mode 100644
index 00000000000..34bfb8dff94
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/SourceProfileSupplier.java
@@ -0,0 +1,34 @@
+/*
+ * 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.streamer;
+
+import org.apache.hudi.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIClass;
+import org.apache.hudi.PublicAPIMethod;
+
+/**
+ * Supplier for SourceProfile
+ */
+@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
+public interface SourceProfileSupplier {
+ @SuppressWarnings("rawtypes")
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ SourceProfile getSourceProfile();
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java
new file mode 100644
index 00000000000..bfe337ee3f2
--- /dev/null
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamContext.java
@@ -0,0 +1,44 @@
+/*
+ * 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.streamer;
+
+import org.apache.hudi.ApiMaturityLevel;
+import org.apache.hudi.PublicAPIClass;
+import org.apache.hudi.PublicAPIMethod;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.schema.SchemaProvider;
+
+/**
+ * The context required to sync one batch of data to hoodie table using
StreamSync.
+ */
+@PublicAPIClass(maturity = ApiMaturityLevel.EVOLVING)
+public interface StreamContext {
+
+ /**
+ * The schema provider used for reading data from source and also writing to
hoodie table.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ SchemaProvider getSchemaProvider();
+
+ /**
+ * An optional stream profile supplying details regarding how the next input
batch in StreamSync should be consumed and written.
+ */
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ Option<SourceProfileSupplier> getSourceProfileSupplier();
+}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
index f87bf083854..4b83ff92b7b 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
@@ -259,19 +259,19 @@ public class StreamSync implements Serializable,
Closeable {
public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession,
SchemaProvider schemaProvider,
TypedProperties props, JavaSparkContext jssc, FileSystem
fs, Configuration conf,
Function<SparkRDDWriteClient, Boolean>
onInitializingHoodieWriteClient) throws IOException {
- this(cfg, sparkSession, schemaProvider, props, new
HoodieSparkEngineContext(jssc), fs, conf, onInitializingHoodieWriteClient);
+ this(cfg, sparkSession, props, new HoodieSparkEngineContext(jssc), fs,
conf, onInitializingHoodieWriteClient, new DefaultStreamContext(schemaProvider,
Option.empty()));
}
- public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession,
SchemaProvider schemaProvider,
+ public StreamSync(HoodieStreamer.Config cfg, SparkSession sparkSession,
TypedProperties props, HoodieSparkEngineContext
hoodieSparkContext, FileSystem fs, Configuration conf,
- Function<SparkRDDWriteClient, Boolean>
onInitializingHoodieWriteClient) throws IOException {
+ Function<SparkRDDWriteClient, Boolean>
onInitializingHoodieWriteClient, StreamContext streamContext) throws
IOException {
this.cfg = cfg;
this.hoodieSparkContext = hoodieSparkContext;
this.sparkSession = sparkSession;
this.fs = fs;
this.onInitializingHoodieWriteClient = onInitializingHoodieWriteClient;
this.props = props;
- this.userProvidedSchemaProvider = schemaProvider;
+ this.userProvidedSchemaProvider = streamContext.getSchemaProvider();
this.processedSchema = new SchemaSet();
this.autoGenerateRecordKeys =
KeyGenUtils.enableAutoGenerateRecordKeys(props);
this.keyGenClassName = getKeyGeneratorClassName(new
TypedProperties(props));
@@ -285,7 +285,7 @@ public class StreamSync implements Serializable, Closeable {
this.errorWriteFailureStrategy =
ErrorTableUtils.getErrorWriteFailureStrategy(props);
}
refreshTimeline();
- Source source = UtilHelpers.createSource(cfg.sourceClassName, props,
hoodieSparkContext.jsc(), sparkSession, schemaProvider, metrics);
+ Source source = UtilHelpers.createSource(cfg.sourceClassName, props,
hoodieSparkContext.jsc(), sparkSession, metrics, streamContext);
this.formatAdapter = new SourceFormatAdapter(source,
this.errorTableWriter, Option.of(props));
Supplier<Option<Schema>> schemaSupplier = schemaProvider == null ?
Option::empty : () -> Option.ofNullable(schemaProvider.getSourceSchema());
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
index b5cbf2738f6..011a1f626b2 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java
@@ -28,6 +28,8 @@ import
org.apache.hudi.utilities.exception.HoodieStreamerException;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
+import org.apache.hudi.utilities.streamer.SourceProfile;
+import org.apache.hudi.utilities.streamer.SourceProfileSupplier;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -52,6 +54,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/**
* Generic tests for all {@link KafkaSource} to ensure all implementations
properly handle offsets, fetch limits, failure modes, etc.
@@ -60,6 +63,7 @@ abstract class BaseTestKafkaSource extends
SparkClientFunctionalTestHarness {
protected static final String TEST_TOPIC_PREFIX = "hoodie_test_";
protected final HoodieIngestionMetrics metrics =
mock(HoodieIngestionMetrics.class);
+ protected final Option<SourceProfileSupplier> sourceProfile =
Option.of(mock(SourceProfileSupplier.class));
protected SchemaProvider schemaProvider;
protected KafkaTestUtils testUtils;
@@ -277,4 +281,51 @@ abstract class BaseTestKafkaSource extends
SparkClientFunctionalTestHarness {
+ " either the data was aged out by Kafka or the topic may have
been deleted before all the data in the topic was processed.",
t.getMessage());
}
+
+ @Test
+ public void testKafkaSourceWithOffsetsFromSourceProfile() {
+ // topic setup.
+ final String topic = TEST_TOPIC_PREFIX + "testKafkaSourceWithOffsetRanges";
+ testUtils.createTopic(topic, 2);
+ TypedProperties props = createPropsForKafkaSource(topic, null, "earliest");
+
+ when(sourceProfile.get().getSourceProfile()).thenReturn(new
TestSourceProfile(Long.MAX_VALUE, 4, 500));
+ SourceFormatAdapter kafkaSource = createSource(props);
+
+ // Test for empty data.
+ assertEquals(Option.empty(),
kafkaSource.fetchNewDataInAvroFormat(Option.empty(),
Long.MAX_VALUE).getBatch());
+
+ // Publish messages and assert source has picked up all messages in
offsetRanges supplied by input batch profile.
+ sendMessagesToKafka(topic, 1000, 2);
+ InputBatch<JavaRDD<GenericRecord>> fetch1 =
kafkaSource.fetchNewDataInAvroFormat(Option.empty(), 900);
+ assertEquals(500, fetch1.getBatch().get().count());
+ }
+
+ static class TestSourceProfile implements SourceProfile<Long> {
+
+ private final long maxSourceBytes;
+ private final int sourcePartitions;
+ private final long numEvents;
+
+ public TestSourceProfile(long maxSourceBytes, int sourcePartitions, long
numEvents) {
+ this.maxSourceBytes = maxSourceBytes;
+ this.sourcePartitions = sourcePartitions;
+ this.numEvents = numEvents;
+ }
+
+ @Override
+ public long getMaxSourceBytes() {
+ return maxSourceBytes;
+ }
+
+ @Override
+ public int getSourcePartitions() {
+ return sourcePartitions;
+ }
+
+ @Override
+ public Long getSourceSpecificContext() {
+ return numEvents;
+ }
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
index 14ffd31582a..59a85a06e9c 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKafkaSource.java
@@ -23,14 +23,16 @@ import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.HoodieAvroRecord;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.testutils.InProcessTimeGenerator;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.InProcessTimeGenerator;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.utilities.UtilHelpers;
import org.apache.hudi.utilities.config.HoodieStreamerConfig;
import org.apache.hudi.utilities.config.KafkaSourceConfig;
import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer;
import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
import org.apache.hudi.utilities.streamer.BaseErrorTableWriter;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.ErrorEvent;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
@@ -60,10 +62,10 @@ import scala.Tuple2;
import static
org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TABLE_BASE_PATH;
import static org.apache.hudi.config.HoodieErrorTableConfig.ERROR_TARGET_TABLE;
import static
org.apache.hudi.utilities.config.KafkaSourceConfig.ENABLE_KAFKA_COMMIT_OFFSET;
+import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_OFFSET_COLUMN;
import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_PARTITION_COLUMN;
import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_TIMESTAMP_COLUMN;
-import static
org.apache.hudi.utilities.schema.KafkaOffsetPostProcessor.KAFKA_SOURCE_KEY_COLUMN;
import static
org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.jsonifyRecords;
import static
org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.jsonifyRecordsByPartitions;
import static
org.apache.hudi.utilities.testutils.UtilitiesTestBase.Helpers.jsonifyRecordsByPartitionsWithNullKafkaKey;
@@ -104,7 +106,7 @@ public class TestJsonKafkaSource extends
BaseTestKafkaSource {
@Override
SourceFormatAdapter createSource(TypedProperties props) {
- return new SourceFormatAdapter(new JsonKafkaSource(props, jsc(), spark(),
schemaProvider, metrics));
+ return new SourceFormatAdapter(new JsonKafkaSource(props, jsc(), spark(),
metrics, new DefaultStreamContext(schemaProvider, sourceProfile)));
}
// test whether empty messages can be filtered
@@ -356,4 +358,13 @@ public class TestJsonKafkaSource extends
BaseTestKafkaSource {
dfWithOffsetInfo.unpersist();
dfWithOffsetInfoAndNullKafkaKey.unpersist();
}
+
+ @Test
+ public void testCreateSource() throws IOException {
+ final String topic = TEST_TOPIC_PREFIX + "testJsonKafkaSourceCreation";
+ testUtils.createTopic(topic, 2);
+ TypedProperties props = createPropsForKafkaSource(topic, null, "earliest");
+ Source jsonKafkaSource =
UtilHelpers.createSource(JsonKafkaSource.class.getName(), props, jsc(),
spark(), metrics, new DefaultStreamContext(schemaProvider, sourceProfile));
+ assertEquals(Source.SourceType.JSON, jsonKafkaSource.getSourceType());
+ }
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
index 52376f89741..b56d87c9263 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestProtoKafkaSource.java
@@ -25,6 +25,7 @@ import org.apache.hudi.utilities.config.KafkaSourceConfig;
import org.apache.hudi.utilities.config.ProtoClassBasedSchemaProviderConfig;
import org.apache.hudi.utilities.schema.ProtoClassBasedSchemaProvider;
import org.apache.hudi.utilities.schema.SchemaProvider;
+import org.apache.hudi.utilities.streamer.DefaultStreamContext;
import org.apache.hudi.utilities.streamer.SourceFormatAdapter;
import org.apache.hudi.utilities.test.proto.Nested;
import org.apache.hudi.utilities.test.proto.Sample;
@@ -89,7 +90,7 @@ public class TestProtoKafkaSource extends BaseTestKafkaSource
{
@Override
SourceFormatAdapter createSource(TypedProperties props) {
this.schemaProvider = new ProtoClassBasedSchemaProvider(props, jsc());
- Source protoKafkaSource = new ProtoKafkaSource(props, jsc(), spark(),
schemaProvider, metrics);
+ Source protoKafkaSource = new ProtoKafkaSource(props, jsc(), spark(),
metrics, new DefaultStreamContext(schemaProvider, sourceProfile));
return new SourceFormatAdapter(protoKafkaSource);
}
