StreamsConfig extends AbstractConfig { public Map<String, Object> getConsumerConfigs(StreamThread streamThread, String groupId, String clientId) { Map<String, Object> props = getBaseConsumerConfigs(); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-consumer"); props.put(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE, streamThread); props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, getInt(REPLICATION_FACTOR_CONFIG)); props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, getInt(NUM_STANDBY_REPLICAS_CONFIG)); props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StreamPartitionAssignor.class.getName()); if (!getString(ZOOKEEPER_CONNECT_CONFIG).equals("")) props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, getString(ZOOKEEPER_CONNECT_CONFIG)); return props; } StreamsConfig(Map<?, ?> props); Map<String, Object> getConsumerConfigs(StreamThread streamThread, String groupId, String clientId); Map<String, Object> getRestoreConsumerConfigs(String clientId); Map<String, Object> getProducerConfigs(String clientId); Serde keySerde(); Serde valueSerde(); static void main(String[] args); static final String STATE_DIR_CONFIG; static final String ZOOKEEPER_CONNECT_CONFIG; static final String COMMIT_INTERVAL_MS_CONFIG; static final String POLL_MS_CONFIG; static final String NUM_STREAM_THREADS_CONFIG; static final String NUM_STANDBY_REPLICAS_CONFIG; static final String BUFFERED_RECORDS_PER_PARTITION_CONFIG; static final String STATE_CLEANUP_DELAY_MS_CONFIG; static final String TIMESTAMP_EXTRACTOR_CLASS_CONFIG; static final String PARTITION_GROUPER_CLASS_CONFIG; static final String APPLICATION_ID_CONFIG; static final String APPLICATION_ID_DOC; static final String REPLICATION_FACTOR_CONFIG; static final String REPLICATION_FACTOR_DOC; static final String KEY_SERDE_CLASS_CONFIG; static final String KEY_SERDE_CLASS_DOC; static final String VALUE_SERDE_CLASS_CONFIG; static final String VALUE_SERDE_CLASS_DOC; static final String METRICS_SAMPLE_WINDOW_MS_CONFIG; static final String METRICS_NUM_SAMPLES_CONFIG; static final String METRIC_REPORTER_CLASSES_CONFIG; static final String BOOTSTRAP_SERVERS_CONFIG; static final String CLIENT_ID_CONFIG; }
@Test public void testGetConsumerConfigs() throws Exception { Map<String, Object> returnedProps = streamsConfig.getConsumerConfigs(null, "example-application", "client"); assertEquals(returnedProps.get(ConsumerConfig.CLIENT_ID_CONFIG), "client-consumer"); assertEquals(returnedProps.get(ConsumerConfig.GROUP_ID_CONFIG), "example-application"); }