EventHubsInputDescriptor extends InputDescriptor<StreamMessageType, EventHubsInputDescriptor<StreamMessageType>> { @Override public Map<String, String> toConfig() { Map<String, String> ehConfigs = new HashMap<>(super.toConfig()); String streamId = getStreamId(); ehConfigs.put(String.format(EventHubConfig.CONFIG_STREAM_NAMESPACE, streamId), namespace); ehConfigs.put(String.format(EventHubConfig.CONFIG_STREAM_ENTITYPATH, streamId), entityPath); sasKeyName.ifPresent(keyName -> ehConfigs.put(String.format(EventHubConfig.CONFIG_STREAM_SAS_KEY_NAME, streamId), keyName)); sasToken.ifPresent(key -> ehConfigs.put(String.format(EventHubConfig.CONFIG_STREAM_SAS_TOKEN, streamId), key)); consumerGroup.ifPresent(consumerGroupName -> ehConfigs.put(String.format(EventHubConfig.CONFIG_STREAM_CONSUMER_GROUP, streamId), consumerGroupName)); return ehConfigs; } EventHubsInputDescriptor(String streamId, String namespace, String entityPath, Serde valueSerde,
      SystemDescriptor systemDescriptor); EventHubsInputDescriptor<StreamMessageType> withSasKeyName(String sasKeyName); EventHubsInputDescriptor<StreamMessageType> withSasKey(String sasToken); EventHubsInputDescriptor<StreamMessageType> withConsumerGroup(String consumerGroup); @Override Map<String, String> toConfig();  }
@Test public void testWithoutEntityConnectionConfigs() { String systemName = "eventHub"; String streamId = "input-stream"; EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor(systemName); EventHubsInputDescriptor<KV<String, String>> inputDescriptor = systemDescriptor .getInputDescriptor(streamId, "entity-namespace", "entity3", new StringSerde()); Map<String, String> generatedConfigs = inputDescriptor.toConfig(); assertEquals("eventHub", generatedConfigs.get("streams.input-stream.samza.system")); assertEquals("entity-namespace", generatedConfigs.get(String.format(EventHubConfig.CONFIG_STREAM_NAMESPACE, streamId))); assertEquals("entity3", generatedConfigs.get(String.format(EventHubConfig.CONFIG_STREAM_ENTITYPATH, streamId))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_STREAM_SAS_KEY_NAME, streamId))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_STREAM_SAS_TOKEN, streamId))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_STREAM_CONSUMER_GROUP, streamId))); assertEquals(3, generatedConfigs.size()); }