@Test public void testGetProducerConfigs() throws Exception { Map<String, Object> returnedProps = streamsConfig.getProducerConfigs("client"); assertEquals(returnedProps.get(ConsumerConfig.CLIENT_ID_CONFIG), "client-producer"); }
public Map<String, Object> getProducerConfigs(String clientId) { Map<String, Object> props = this.originals(); props.remove(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG); removeStreamsSpecificConfigs(props); props.put(ProducerConfig.LINGER_MS_CONFIG, "100"); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(String clientId) { Map<String, Object> props = this.originals(); props.remove(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG); removeStreamsSpecificConfigs(props); props.put(ProducerConfig.LINGER_MS_CONFIG, "100"); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; } }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(String clientId) { Map<String, Object> props = this.originals(); props.remove(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG); removeStreamsSpecificConfigs(props); props.put(ProducerConfig.LINGER_MS_CONFIG, "100"); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; } StreamsConfig(Map<?, ?> props); }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(String clientId) { Map<String, Object> props = this.originals(); props.remove(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG); removeStreamsSpecificConfigs(props); props.put(ProducerConfig.LINGER_MS_CONFIG, "100"); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); 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); }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(String clientId) { Map<String, Object> props = this.originals(); props.remove(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG); removeStreamsSpecificConfigs(props); props.put(ProducerConfig.LINGER_MS_CONFIG, "100"); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); 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; }