[incubator-gobblin] branch master updated: [GOBBLIN-761] Only instantiate topic-specific configStore object when topic.name is available

[hu...@apache.org]

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

hutran 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 51ccc8d  [GOBBLIN-761] Only instantiate topic-specific configStore object when topic.name is available
51ccc8d is described below

commit 51ccc8df26297aa59231dcce4f6c053d07af4656
Author: autumnust <le...@linkedin.com>
AuthorDate: Wed May 1 17:21:42 2019 -0700

    [GOBBLIN-761] Only instantiate topic-specific configStore object when topic.name is available
    
    Closes #2625 from autumnust/master
---
 .../apache/gobblin/hive/policy/HiveRegistrationPolicyBase.java | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/policy/HiveRegistrationPolicyBase.java b/gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/policy/HiveRegistrationPolicyBase.java
index ee1eb03..ca4d35b 100644
--- a/gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/policy/HiveRegistrationPolicyBase.java
+++ b/gobblin-hive-registration/src/main/java/org/apache/gobblin/hive/policy/HiveRegistrationPolicyBase.java
@@ -137,10 +137,12 @@ public class HiveRegistrationPolicyBase implements HiveRegistrationPolicy {
 
     // Get Topic-specific config object doesn't require any runtime-set properties in prop object, safe to initialize
     // in constructor.
-    Timer.Context context = this.metricContext.timer(CONFIG_FOR_TOPIC_TIMER).time();
-    configForTopic =
-        ConfigStoreUtils.getConfigForTopic(this.props.getProperties(), KafkaSource.TOPIC_NAME, this.configClient);
-    context.close();
+    if (this.props.getProperties().containsKey(KafkaSource.TOPIC_NAME)) {
+      Timer.Context context = this.metricContext.timer(CONFIG_FOR_TOPIC_TIMER).time();
+      configForTopic =
+          ConfigStoreUtils.getConfigForTopic(this.props.getProperties(), KafkaSource.TOPIC_NAME, this.configClient);
+      context.close();
+    }
   }
 
   /**