@Test public void testGetDatasetSpecificStateWithRegex() { String dataset1 = "testDataset1"; String dataset2 = "testDataset2"; String dataset3 = "otherTestDataset1"; String testKey1 = "testKey1"; String testValue1 = "testValue1"; SourceState state = new SourceState(); state.setProp(DatasetUtils.DATASET_SPECIFIC_PROPS, "[{\"dataset\" : \"testDataset.*\", \"" + testKey1 + "\" : \"" + testValue1 + "\"}]"); Map<String, State> datasetSpecificStateMap = DatasetUtils.getDatasetSpecificProps(Lists.newArrayList(dataset1, dataset2, dataset3), state); State dataset1ExpectedState = new State(); dataset1ExpectedState.setProp(testKey1, testValue1); State dataset2ExpectedState = new State(); dataset2ExpectedState.setProp(testKey1, testValue1); Assert.assertEquals(datasetSpecificStateMap.get(dataset1), dataset1ExpectedState); Assert.assertEquals(datasetSpecificStateMap.get(dataset2), dataset2ExpectedState); Assert.assertNull(datasetSpecificStateMap.get(dataset3)); }
public static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state) { if (!Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) || !Strings.isNullOrEmpty(state.getProp(KAFKA_TOPIC_SPECIFIC_STATE))) { Map<String, State> datasetSpecificConfigMap = Maps.newHashMap(); JsonArray array = !Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) ? state.getPropAsJsonArray(DATASET_SPECIFIC_PROPS) : state.getPropAsJsonArray(KAFKA_TOPIC_SPECIFIC_STATE); for (JsonElement datasetElement : array) { Preconditions.checkArgument(datasetElement.isJsonObject(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed"); JsonObject object = datasetElement.getAsJsonObject(); if (object.has(DATASET)) { JsonElement datasetNameElement = object.get(DATASET); Preconditions.checkArgument(datasetNameElement.isJsonPrimitive(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed, the " + DATASET + " field must be a string"); for (String dataset : Iterables.filter(datasets, new DatasetPredicate(datasetNameElement.getAsString()))) { if (datasetSpecificConfigMap.containsKey(dataset)) { datasetSpecificConfigMap.get(dataset).addAll(StateUtils.jsonObjectToState(object, DATASET)); } else { datasetSpecificConfigMap.put(dataset, StateUtils.jsonObjectToState(object, DATASET)); } } } else { LOG.warn("Skipping JsonElement " + datasetElement + " as it is does not contain a field with key " + DATASET); } } return datasetSpecificConfigMap; } return Maps.newHashMap(); }
DatasetUtils { public static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state) { if (!Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) || !Strings.isNullOrEmpty(state.getProp(KAFKA_TOPIC_SPECIFIC_STATE))) { Map<String, State> datasetSpecificConfigMap = Maps.newHashMap(); JsonArray array = !Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) ? state.getPropAsJsonArray(DATASET_SPECIFIC_PROPS) : state.getPropAsJsonArray(KAFKA_TOPIC_SPECIFIC_STATE); for (JsonElement datasetElement : array) { Preconditions.checkArgument(datasetElement.isJsonObject(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed"); JsonObject object = datasetElement.getAsJsonObject(); if (object.has(DATASET)) { JsonElement datasetNameElement = object.get(DATASET); Preconditions.checkArgument(datasetNameElement.isJsonPrimitive(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed, the " + DATASET + " field must be a string"); for (String dataset : Iterables.filter(datasets, new DatasetPredicate(datasetNameElement.getAsString()))) { if (datasetSpecificConfigMap.containsKey(dataset)) { datasetSpecificConfigMap.get(dataset).addAll(StateUtils.jsonObjectToState(object, DATASET)); } else { datasetSpecificConfigMap.put(dataset, StateUtils.jsonObjectToState(object, DATASET)); } } } else { LOG.warn("Skipping JsonElement " + datasetElement + " as it is does not contain a field with key " + DATASET); } } return datasetSpecificConfigMap; } return Maps.newHashMap(); } }
DatasetUtils { public static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state) { if (!Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) || !Strings.isNullOrEmpty(state.getProp(KAFKA_TOPIC_SPECIFIC_STATE))) { Map<String, State> datasetSpecificConfigMap = Maps.newHashMap(); JsonArray array = !Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) ? state.getPropAsJsonArray(DATASET_SPECIFIC_PROPS) : state.getPropAsJsonArray(KAFKA_TOPIC_SPECIFIC_STATE); for (JsonElement datasetElement : array) { Preconditions.checkArgument(datasetElement.isJsonObject(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed"); JsonObject object = datasetElement.getAsJsonObject(); if (object.has(DATASET)) { JsonElement datasetNameElement = object.get(DATASET); Preconditions.checkArgument(datasetNameElement.isJsonPrimitive(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed, the " + DATASET + " field must be a string"); for (String dataset : Iterables.filter(datasets, new DatasetPredicate(datasetNameElement.getAsString()))) { if (datasetSpecificConfigMap.containsKey(dataset)) { datasetSpecificConfigMap.get(dataset).addAll(StateUtils.jsonObjectToState(object, DATASET)); } else { datasetSpecificConfigMap.put(dataset, StateUtils.jsonObjectToState(object, DATASET)); } } } else { LOG.warn("Skipping JsonElement " + datasetElement + " as it is does not contain a field with key " + DATASET); } } return datasetSpecificConfigMap; } return Maps.newHashMap(); } private  DatasetUtils(); }
DatasetUtils { public static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state) { if (!Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) || !Strings.isNullOrEmpty(state.getProp(KAFKA_TOPIC_SPECIFIC_STATE))) { Map<String, State> datasetSpecificConfigMap = Maps.newHashMap(); JsonArray array = !Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) ? state.getPropAsJsonArray(DATASET_SPECIFIC_PROPS) : state.getPropAsJsonArray(KAFKA_TOPIC_SPECIFIC_STATE); for (JsonElement datasetElement : array) { Preconditions.checkArgument(datasetElement.isJsonObject(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed"); JsonObject object = datasetElement.getAsJsonObject(); if (object.has(DATASET)) { JsonElement datasetNameElement = object.get(DATASET); Preconditions.checkArgument(datasetNameElement.isJsonPrimitive(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed, the " + DATASET + " field must be a string"); for (String dataset : Iterables.filter(datasets, new DatasetPredicate(datasetNameElement.getAsString()))) { if (datasetSpecificConfigMap.containsKey(dataset)) { datasetSpecificConfigMap.get(dataset).addAll(StateUtils.jsonObjectToState(object, DATASET)); } else { datasetSpecificConfigMap.put(dataset, StateUtils.jsonObjectToState(object, DATASET)); } } } else { LOG.warn("Skipping JsonElement " + datasetElement + " as it is does not contain a field with key " + DATASET); } } return datasetSpecificConfigMap; } return Maps.newHashMap(); } private  DatasetUtils(); static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state); }
DatasetUtils { public static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state) { if (!Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) || !Strings.isNullOrEmpty(state.getProp(KAFKA_TOPIC_SPECIFIC_STATE))) { Map<String, State> datasetSpecificConfigMap = Maps.newHashMap(); JsonArray array = !Strings.isNullOrEmpty(state.getProp(DATASET_SPECIFIC_PROPS)) ? state.getPropAsJsonArray(DATASET_SPECIFIC_PROPS) : state.getPropAsJsonArray(KAFKA_TOPIC_SPECIFIC_STATE); for (JsonElement datasetElement : array) { Preconditions.checkArgument(datasetElement.isJsonObject(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed"); JsonObject object = datasetElement.getAsJsonObject(); if (object.has(DATASET)) { JsonElement datasetNameElement = object.get(DATASET); Preconditions.checkArgument(datasetNameElement.isJsonPrimitive(), "The value for property " + DATASET_SPECIFIC_PROPS + " is malformed, the " + DATASET + " field must be a string"); for (String dataset : Iterables.filter(datasets, new DatasetPredicate(datasetNameElement.getAsString()))) { if (datasetSpecificConfigMap.containsKey(dataset)) { datasetSpecificConfigMap.get(dataset).addAll(StateUtils.jsonObjectToState(object, DATASET)); } else { datasetSpecificConfigMap.put(dataset, StateUtils.jsonObjectToState(object, DATASET)); } } } else { LOG.warn("Skipping JsonElement " + datasetElement + " as it is does not contain a field with key " + DATASET); } } return datasetSpecificConfigMap; } return Maps.newHashMap(); } private  DatasetUtils(); static Map<String, State> getDatasetSpecificProps(Iterable<String> datasets, State state); static final String DATASET; static final String DATASET_SPECIFIC_PROPS; }