EventHubsSystemDescriptor extends SystemDescriptor<EventHubsSystemDescriptor> { @Override public Map<String, String> toConfig() { Map<String, String> ehConfigs = new HashMap<>(super.toConfig()); String systemName = getSystemName(); if (!this.streamIds.isEmpty()) { ehConfigs.put(String.format(EventHubConfig.CONFIG_STREAM_LIST, systemName), String.join(",", this.streamIds)); } this.fetchRuntimeInfoTimeout.ifPresent(timeout -> ehConfigs.put(String.format(EventHubConfig.CONFIG_FETCH_RUNTIME_INFO_TIMEOUT_MILLIS, systemName), Integer.toString(timeout))); this.numClientThreads.ifPresent(numClientThreads -> ehConfigs.put(String.format(EventHubConfig.CONFIG_SYSTEM_NUM_CLIENT_THREADS, systemName), Integer.toString(numClientThreads))); this.consumerReceiveQueueSize.ifPresent(receiveQueueSize -> ehConfigs.put(String.format(EventHubConfig.CONFIG_CONSUMER_BUFFER_CAPACITY, systemName), Integer.toString(receiveQueueSize))); this.consumerMaxEventCountPerPoll.ifPresent(count -> ehConfigs.put(String.format(EventHubConfig.CONFIG_MAX_EVENT_COUNT_PER_POLL, systemName), Integer.toString(count))); this.consumerPrefetchCount.ifPresent(count -> ehConfigs.put(String.format(EventHubConfig.CONFIG_PREFETCH_COUNT, systemName), Integer.toString(count))); this.producerEventhubsSendKey.ifPresent(sendKeys -> ehConfigs.put(String.format(EventHubConfig.CONFIG_SEND_KEY_IN_EVENT_PROPERTIES, systemName), Boolean.toString(sendKeys))); this.producerEventhubsPartitioningMethod.ifPresent(partitioningMethod -> ehConfigs.put(String.format(EventHubConfig.CONFIG_PRODUCER_PARTITION_METHOD, systemName), partitioningMethod.toString())); return ehConfigs; } EventHubsSystemDescriptor(String systemName); EventHubsInputDescriptor<KV<String, ValueType>> getInputDescriptor(String streamId, String namespace,
      String entityPath, Serde<ValueType> valueSerde); EventHubsOutputDescriptor<KV<String, ValueType>> getOutputDescriptor(String streamId, String namespace,
      String entityPath, Serde<ValueType> valueSerde); EventHubsSystemDescriptor withRuntimeInfoTimeout(int timeoutMS); EventHubsSystemDescriptor withNumClientThreads(int numClientThreads); EventHubsSystemDescriptor withReceiveQueueSize(int receiveQueueSize); EventHubsSystemDescriptor withMaxEventCountPerPoll(int count); EventHubsSystemDescriptor withPrefetchCount(int count); EventHubsSystemDescriptor withPartitioningMethod(PartitioningMethod partitioningMethod); EventHubsSystemDescriptor withSendKeys(boolean sendKeys); @Override Map<String, String> toConfig();  }
@Test public void testWithoutDescriptorOverrides() { String systemName = "eventHub"; EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor(systemName); Map<String, String> generatedConfigs = systemDescriptor.toConfig(); assertEquals("org.apache.samza.system.eventhub.EventHubSystemFactory", generatedConfigs.get(String.format("systems.%s.samza.factory", systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_MAX_EVENT_COUNT_PER_POLL, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_SYSTEM_NUM_CLIENT_THREADS, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_PRODUCER_PARTITION_METHOD, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_PREFETCH_COUNT, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_CONSUMER_BUFFER_CAPACITY, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_FETCH_RUNTIME_INFO_TIMEOUT_MILLIS, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_SEND_KEY_IN_EVENT_PROPERTIES, systemName))); assertNull(generatedConfigs.get(String.format(EventHubConfig.CONFIG_STREAM_LIST, systemName))); assertEquals(1, generatedConfigs.size()); }