@Test public void testGetBrokerConfig() { Map<String, String> srcConfigMap = new HashMap<>(); srcConfigMap.put("gobblin.broker.key1", "value1"); Config brokerConfig = SharedResourcesBrokerFactory.getBrokerConfig(ConfigFactory.parseMap(srcConfigMap)); Config expectedConfig = ConfigFactory.parseMap(ImmutableMap.of("key1", "value1")); Assert.assertEquals(brokerConfig, expectedConfig); srcConfigMap.put("gobblin.shared.key2", "value2"); srcConfigMap.put("gobblin.brokerNamespaces", "gobblin.shared"); brokerConfig = SharedResourcesBrokerFactory.getBrokerConfig(ConfigFactory.parseMap(srcConfigMap)); expectedConfig = ConfigFactory.parseMap(ImmutableMap.of("key1", "value1","key2", "value2")); Assert.assertEquals(brokerConfig, expectedConfig); srcConfigMap.put("gobblin.shared.key2", "value2"); srcConfigMap.put("gobblin.shared.key3", "value3"); srcConfigMap.put("gobblin.shared2.key3", "value3x"); srcConfigMap.put("gobblin.shared2.key4", "value4"); srcConfigMap.put("gobblin.brokerNamespaces", "gobblin.shared, gobblin.shared2"); brokerConfig = SharedResourcesBrokerFactory.getBrokerConfig(ConfigFactory.parseMap(srcConfigMap)); expectedConfig = ConfigFactory.parseMap(ImmutableMap.of("key1", "value1", "key2", "value2", "key3", "value3", "key4", "value4")); Assert.assertEquals(brokerConfig, expectedConfig); }
@VisibleForTesting static Config getBrokerConfig(Config srcConfig) { Config allSrcConfig = srcConfig.withFallback(BROKER_NAMESPACES_FALLBACK); String namespaces = allSrcConfig.getString(BrokerConstants.GOBBLIN_BROKER_CONFIG_NAMESPACES); Config brokerConfig = ConfigUtils.getConfigOrEmpty(allSrcConfig, BrokerConstants.GOBBLIN_BROKER_CONFIG_PREFIX); for (String namespace : LIST_SPLITTER.splitToList(namespaces)) { brokerConfig = brokerConfig.withFallback(ConfigUtils.getConfigOrEmpty(allSrcConfig, namespace)); } return brokerConfig; }
SharedResourcesBrokerFactory { @VisibleForTesting static Config getBrokerConfig(Config srcConfig) { Config allSrcConfig = srcConfig.withFallback(BROKER_NAMESPACES_FALLBACK); String namespaces = allSrcConfig.getString(BrokerConstants.GOBBLIN_BROKER_CONFIG_NAMESPACES); Config brokerConfig = ConfigUtils.getConfigOrEmpty(allSrcConfig, BrokerConstants.GOBBLIN_BROKER_CONFIG_PREFIX); for (String namespace : LIST_SPLITTER.splitToList(namespaces)) { brokerConfig = brokerConfig.withFallback(ConfigUtils.getConfigOrEmpty(allSrcConfig, namespace)); } return brokerConfig; } }
SharedResourcesBrokerFactory { @VisibleForTesting static Config getBrokerConfig(Config srcConfig) { Config allSrcConfig = srcConfig.withFallback(BROKER_NAMESPACES_FALLBACK); String namespaces = allSrcConfig.getString(BrokerConstants.GOBBLIN_BROKER_CONFIG_NAMESPACES); Config brokerConfig = ConfigUtils.getConfigOrEmpty(allSrcConfig, BrokerConstants.GOBBLIN_BROKER_CONFIG_PREFIX); for (String namespace : LIST_SPLITTER.splitToList(namespaces)) { brokerConfig = brokerConfig.withFallback(ConfigUtils.getConfigOrEmpty(allSrcConfig, namespace)); } return brokerConfig; }  }
SharedResourcesBrokerFactory { @VisibleForTesting static Config getBrokerConfig(Config srcConfig) { Config allSrcConfig = srcConfig.withFallback(BROKER_NAMESPACES_FALLBACK); String namespaces = allSrcConfig.getString(BrokerConstants.GOBBLIN_BROKER_CONFIG_NAMESPACES); Config brokerConfig = ConfigUtils.getConfigOrEmpty(allSrcConfig, BrokerConstants.GOBBLIN_BROKER_CONFIG_PREFIX); for (String namespace : LIST_SPLITTER.splitToList(namespaces)) { brokerConfig = brokerConfig.withFallback(ConfigUtils.getConfigOrEmpty(allSrcConfig, namespace)); } return brokerConfig; }  static SharedResourcesBrokerImpl<S> createDefaultTopLevelBroker(Config config,
      ScopeInstance<S> globalScope); static SharedResourcesBroker<?> getImplicitBroker(); static void registerImplicitBroker(SharedResourcesBroker<?> broker); static void addBrokerKeys(Map<String, String> configMap, Iterable<Map.Entry<S, T>> entries); }
SharedResourcesBrokerFactory { @VisibleForTesting static Config getBrokerConfig(Config srcConfig) { Config allSrcConfig = srcConfig.withFallback(BROKER_NAMESPACES_FALLBACK); String namespaces = allSrcConfig.getString(BrokerConstants.GOBBLIN_BROKER_CONFIG_NAMESPACES); Config brokerConfig = ConfigUtils.getConfigOrEmpty(allSrcConfig, BrokerConstants.GOBBLIN_BROKER_CONFIG_PREFIX); for (String namespace : LIST_SPLITTER.splitToList(namespaces)) { brokerConfig = brokerConfig.withFallback(ConfigUtils.getConfigOrEmpty(allSrcConfig, namespace)); } return brokerConfig; }  static SharedResourcesBrokerImpl<S> createDefaultTopLevelBroker(Config config,
      ScopeInstance<S> globalScope); static SharedResourcesBroker<?> getImplicitBroker(); static void registerImplicitBroker(SharedResourcesBroker<?> broker); static void addBrokerKeys(Map<String, String> configMap, Iterable<Map.Entry<S, T>> entries); static final String LOAD_HADOOP_CONFIGURATION; static final String BROKER_CONF_FILE_KEY; static final String DEFAULT_BROKER_CONF_FILE; }