Generialize UniversalKafkaSource to accept Extractor that not extending KafkaExtractor
Closes from autumnust/KafkaExtractor
@@ -19,19 +19,17 @@ package org.apache.gobblin.source.extractor.extract.kafka;
import java.io.IOException;
-import com.google.common.base.Preconditions;
-
import org.apache.gobblin.configuration.WorkUnitState;
import org.apache.gobblin.source.extractor.Extractor;
+import org.apache.gobblin.source.extractor.extract.EventBasedExtractor;
import org.apache.gobblin.util.ClassAliasResolver;
import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
+import com.google.common.base.Preconditions;
/**
- * A {@link KafkaSource} to use with arbitrary {@link KafkaExtractor}. Specify the extractor to use with key
+ * A {@link KafkaSource} to use with arbitrary {@link EventBasedExtractor}. Specify the extractor to use with key
* {@link #EXTRACTOR_TYPE}.
*/
public class UniversalKafkaSource<S, D> extends KafkaSource<S, D> {
@@ -39,12 +37,13 @@ public class UniversalKafkaSource<S, D> extends KafkaSource<S, D> {
public static final String EXTRACTOR_TYPE = "gobblin.source.kafka.extractorType";
@Override
- public Extractor<S, D> getExtractor(WorkUnitState state) throws IOException {
+ public Extractor<S, D> getExtractor(WorkUnitState state)
+ throws IOException {
Preconditions.checkArgument(state.contains(EXTRACTOR_TYPE), "Missing key " + EXTRACTOR_TYPE);
try {
- ClassAliasResolver<KafkaExtractor> aliasResolver = new ClassAliasResolver<>(KafkaExtractor.class);
- Class<? extends KafkaExtractor> klazz = aliasResolver.resolveClass(state.getProp(EXTRACTOR_TYPE));
+ ClassAliasResolver<EventBasedExtractor> aliasResolver = new ClassAliasResolver<>(EventBasedExtractor.class);
+ Class<? extends EventBasedExtractor> klazz = aliasResolver.resolveClass(state.getProp(EXTRACTOR_TYPE));
return GobblinConstructorUtils.invokeLongestConstructor(klazz, state);
} catch (ReflectiveOperationException e) {
