KafkaInputDescriptor extends InputDescriptor<StreamMessageType, KafkaInputDescriptor<StreamMessageType>> { @Override public Map<String, String> toConfig() { Map<String, String> configs = new HashMap<>(super.toConfig()); String streamName = getPhysicalName().orElse(getStreamId()); String systemName = getSystemName(); consumerAutoOffsetResetOptional.ifPresent(autoOffsetReset -> configs.put(String.format(CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY, systemName, streamName), autoOffsetReset)); consumerFetchMessageMaxBytesOptional.ifPresent(fetchMessageMaxBytes -> configs.put(String.format(CONSUMER_FETCH_MESSAGE_MAX_BYTES_CONFIG_KEY, systemName, streamName), Long.toString(fetchMessageMaxBytes))); return configs; } KafkaInputDescriptor(String streamId, SystemDescriptor systemDescriptor, Serde serde, InputTransformer transformer); KafkaInputDescriptor<StreamMessageType> withConsumerAutoOffsetReset(String consumerAutoOffsetReset); KafkaInputDescriptor<StreamMessageType> withConsumerFetchMessageMaxBytes(long fetchMessageMaxBytes); @Override Map<String, String> toConfig();  }
@Test public void testISDConfigsWithDefaults() { KafkaSystemDescriptor sd = new KafkaSystemDescriptor("kafka") .withConsumerZkConnect(ImmutableList.of("localhost:123")) .withProducerBootstrapServers(ImmutableList.of("localhost:567", "localhost:890")); KafkaInputDescriptor<KV<String, Integer>> isd = sd.getInputDescriptor("input-stream", KVSerde.of(new StringSerde(), new IntegerSerde())); Map<String, String> generatedConfigs = isd.toConfig(); assertEquals("kafka", generatedConfigs.get("streams.input-stream.samza.system")); assertEquals(1, generatedConfigs.size()); }