KafkaSystemDescriptor extends SystemDescriptor<KafkaSystemDescriptor> implements SimpleInputDescriptorProvider, OutputDescriptorProvider { @Override public Map<String, String> toConfig() { Map<String, String> configs = new HashMap<>(super.toConfig()); if (!consumerZkConnect.isEmpty()) { configs.put(String.format(CONSUMER_ZK_CONNECT_CONFIG_KEY, getSystemName()), String.join(",", consumerZkConnect)); } consumerAutoOffsetResetOptional.ifPresent(consumerAutoOffsetReset -> configs.put(String.format(CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY, getSystemName()), consumerAutoOffsetReset)); consumerFetchThresholdOptional.ifPresent(consumerFetchThreshold -> configs.put(String.format(CONSUMER_FETCH_THRESHOLD_CONFIG_KEY, getSystemName()), Integer.toString(consumerFetchThreshold))); consumerFetchThresholdBytesOptional.ifPresent(consumerFetchThresholdBytes -> configs.put(String.format(CONSUMER_FETCH_THRESHOLD_BYTES_CONFIG_KEY, getSystemName()), Long.toString(consumerFetchThresholdBytes))); consumerFetchMessageMaxBytesOptional.ifPresent(consumerFetchMessageMaxBytes -> configs.put(String.format(CONSUMER_FETCH_MESSAGE_MAX_BYTES_KEY, getSystemName()), Long.toString(consumerFetchMessageMaxBytes))); consumerConfigs.forEach((key, value) -> configs.put(String.format(CONSUMER_CONFIGS_CONFIG_KEY, getSystemName(), key), value)); if (!producerBootstrapServers.isEmpty()) { configs.put(String.format(PRODUCER_BOOTSTRAP_SERVERS_CONFIG_KEY, getSystemName()), String.join(",", producerBootstrapServers)); } producerConfigs.forEach((key, value) -> configs.put(String.format(PRODUCER_CONFIGS_CONFIG_KEY, getSystemName(), key), value)); return configs; } KafkaSystemDescriptor(String systemName); @Override KafkaInputDescriptor<StreamMessageType> getInputDescriptor(String streamId, Serde<StreamMessageType> serde); @Override KafkaOutputDescriptor<StreamMessageType> getOutputDescriptor(String streamId, Serde<StreamMessageType> serde); KafkaSystemDescriptor withConsumerZkConnect(List<String> consumerZkConnect); KafkaSystemDescriptor withConsumerAutoOffsetReset(String consumerAutoOffsetReset); KafkaSystemDescriptor withSamzaFetchThreshold(int fetchThreshold); KafkaSystemDescriptor withSamzaFetchThresholdBytes(long fetchThresholdBytes); KafkaSystemDescriptor withConsumerFetchMessageMaxBytes(long fetchMessageMaxBytes); KafkaSystemDescriptor withConsumerConfigs(Map<String, String> consumerConfigs); KafkaSystemDescriptor withProducerBootstrapServers(List<String> producerBootstrapServers); KafkaSystemDescriptor withProducerConfigs(Map<String, String> producerConfigs); @Override Map<String, String> toConfig();  }
@Test public void testSDConfigsWithoutOverrides() { KafkaSystemDescriptor sd = new KafkaSystemDescriptor("kafka"); Map<String, String> generatedConfigs = sd.toConfig(); assertEquals("org.apache.samza.system.kafka.KafkaSystemFactory", generatedConfigs.get("systems.kafka.samza.factory")); assertEquals(1, generatedConfigs.size()); }