@Test public void shouldAllowSettingProducerEnableIdempotenceIfEosDisabled() { props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); final StreamsConfig streamsConfig = new StreamsConfig(props); streamsConfig.getProducerConfigs("clientId"); }
public Map<String, Object> getProducerConfigs(final String clientId) { final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames()); if (eosEnabled) { if (clientProvidedProps.containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have idempotency enabled."); } if (clientProvidedProps.containsKey(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have only one in-flight request per connection."); } } final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES); props.putAll(clientProvidedProps); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG)); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(final String clientId) { final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames()); if (eosEnabled) { if (clientProvidedProps.containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have idempotency enabled."); } if (clientProvidedProps.containsKey(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have only one in-flight request per connection."); } } final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES); props.putAll(clientProvidedProps); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG)); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; } }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(final String clientId) { final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames()); if (eosEnabled) { if (clientProvidedProps.containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have idempotency enabled."); } if (clientProvidedProps.containsKey(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have only one in-flight request per connection."); } } final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES); props.putAll(clientProvidedProps); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG)); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; } StreamsConfig(final Map<?, ?> props); }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(final String clientId) { final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames()); if (eosEnabled) { if (clientProvidedProps.containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have idempotency enabled."); } if (clientProvidedProps.containsKey(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have only one in-flight request per connection."); } } final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES); props.putAll(clientProvidedProps); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG)); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; } StreamsConfig(final Map<?, ?> props); static String consumerPrefix(final String consumerProp); static String producerPrefix(final String producerProp); static ConfigDef configDef(); Map<String, Object> getConsumerConfigs(final StreamThread streamThread,
                                                  final String groupId,
                                                  final String clientId); Map<String, Object> getRestoreConsumerConfigs(final String clientId); Map<String, Object> getProducerConfigs(final String clientId); @Deprecated Serde keySerde(); Serde defaultKeySerde(); @Deprecated Serde valueSerde(); Serde defaultValueSerde(); TimestampExtractor defaultTimestampExtractor(); static void main(final String[] args); }
StreamsConfig extends AbstractConfig { public Map<String, Object> getProducerConfigs(final String clientId) { final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(PRODUCER_PREFIX, ProducerConfig.configNames()); if (eosEnabled) { if (clientProvidedProps.containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have idempotency enabled."); } if (clientProvidedProps.containsKey(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { throw new ConfigException("Unexpected user-specified consumer config " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + "; because " + PROCESSING_GUARANTEE_CONFIG + " is set to '" + EXACTLY_ONCE + "' producer will always have only one in-flight request per connection."); } } final Map<String, Object> props = new HashMap<>(eosEnabled ? PRODUCER_EOS_OVERRIDES : PRODUCER_DEFAULT_OVERRIDES); props.putAll(clientProvidedProps); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, originals().get(BOOTSTRAP_SERVERS_CONFIG)); props.put(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer"); return props; } StreamsConfig(final Map<?, ?> props); static String consumerPrefix(final String consumerProp); static String producerPrefix(final String producerProp); static ConfigDef configDef(); Map<String, Object> getConsumerConfigs(final StreamThread streamThread,
                                                  final String groupId,
                                                  final String clientId); Map<String, Object> getRestoreConsumerConfigs(final String clientId); Map<String, Object> getProducerConfigs(final String clientId); @Deprecated Serde keySerde(); Serde defaultKeySerde(); @Deprecated Serde valueSerde(); Serde defaultValueSerde(); TimestampExtractor defaultTimestampExtractor(); static void main(final String[] args); static final String CONSUMER_PREFIX; static final String PRODUCER_PREFIX; static final String AT_LEAST_ONCE; static final String EXACTLY_ONCE; static final String APPLICATION_ID_CONFIG; static final String APPLICATION_SERVER_CONFIG; static final String BOOTSTRAP_SERVERS_CONFIG; static final String BUFFERED_RECORDS_PER_PARTITION_CONFIG; static final String CACHE_MAX_BYTES_BUFFERING_CONFIG; static final String CLIENT_ID_CONFIG; static final String COMMIT_INTERVAL_MS_CONFIG; static final String CONNECTIONS_MAX_IDLE_MS_CONFIG; static final String DEFAULT_KEY_SERDE_CLASS_CONFIG; static final String DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG; static final String DEFAULT_VALUE_SERDE_CLASS_CONFIG; @Deprecated
    static final String KEY_SERDE_CLASS_CONFIG; static final String METADATA_MAX_AGE_CONFIG; static final String METRICS_NUM_SAMPLES_CONFIG; static final String METRICS_RECORDING_LEVEL_CONFIG; static final String METRIC_REPORTER_CLASSES_CONFIG; static final String METRICS_SAMPLE_WINDOW_MS_CONFIG; static final String NUM_STANDBY_REPLICAS_CONFIG; static final String NUM_STREAM_THREADS_CONFIG; static final String PARTITION_GROUPER_CLASS_CONFIG; static final String POLL_MS_CONFIG; static final String PROCESSING_GUARANTEE_CONFIG; static final String RECEIVE_BUFFER_CONFIG; static final String RECONNECT_BACKOFF_MS_CONFIG; static final String RECONNECT_BACKOFF_MAX_MS_CONFIG; static final String REPLICATION_FACTOR_CONFIG; static final String REQUEST_TIMEOUT_MS_CONFIG; static final String RETRY_BACKOFF_MS_CONFIG; static final String ROCKSDB_CONFIG_SETTER_CLASS_CONFIG; static final String SECURITY_PROTOCOL_CONFIG; static final String SEND_BUFFER_CONFIG; static final String STATE_CLEANUP_DELAY_MS_CONFIG; static final String STATE_DIR_CONFIG; @Deprecated
    static final String TIMESTAMP_EXTRACTOR_CLASS_CONFIG; @Deprecated
    static final String VALUE_SERDE_CLASS_CONFIG; static final String WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG; @Deprecated
    static final String ZOOKEEPER_CONNECT_CONFIG; }