[incubator-gobblin] branch master updated: [GOBBLIN-1356] Make hive registration in compaction also be able to pick up topic config in configStore

[lesun]

This is an automated email from the ASF dual-hosted git repository.

lesun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new f26db67  [GOBBLIN-1356] Make hive registration in compaction also be 
able to pick up topic config in configStore
f26db67 is described below

commit f26db677a16db058767c39757a2d1ce43c89cdd2
Author: Zihan Li <zi...@zihli-mn1.linkedin.biz>
AuthorDate: Fri Jan 15 11:52:40 2021 -0800

    [GOBBLIN-1356] Make hive registration in compaction also be able to pick up 
topic config in configStore
    
    remove assumption that we only support 2 types in
    union
    
    address typos
    
    set the topic name, so that hive registration in
    compaction can pick config in config store
    
    Closes #3197 from ZihanLi58/GOBBLIN-1356
---
 .../gobblin/compaction/action/CompactionHiveRegistrationAction.java     | 2 ++
 1 file changed, 2 insertions(+)

diff --git 
a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionHiveRegistrationAction.java
 
b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionHiveRegistrationAction.java
index b5c3ca1..c0147fb 100644
--- 
a/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionHiveRegistrationAction.java
+++ 
b/gobblin-compaction/src/main/java/org/apache/gobblin/compaction/action/CompactionHiveRegistrationAction.java
@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 
 import org.apache.gobblin.compaction.verify.InputRecordCountHelper;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaSource;
 import org.apache.gobblin.util.PathUtils;
 import org.apache.hadoop.fs.Path;
 
@@ -92,6 +93,7 @@ public class CompactionHiveRegistrationAction implements 
CompactionCompleteActio
     }
 
     try (HiveRegister hiveRegister = HiveRegister.get(state)) {
+      state.setProp(KafkaSource.TOPIC_NAME, result.getDatasetName());
       HiveRegistrationPolicy hiveRegistrationPolicy = 
HiveRegistrationPolicyBase.getPolicy(state);
 
       List<String> paths = new ArrayList<>();