KafkaBinderConfigurationProperties { public Map<String, Object> mergedConsumerConfiguration() { Map<String, Object> consumerConfiguration = new HashMap<>(); consumerConfiguration.putAll(this.kafkaProperties.buildConsumerProperties()); for (Map.Entry<String, String> configurationEntry : this.configuration .entrySet()) { if (ConsumerConfig.configNames().contains(configurationEntry.getKey())) { consumerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue()); } } consumerConfiguration.putAll(this.consumerProperties); filterStreamManagedConfiguration(consumerConfiguration); return getConfigurationWithBootstrapServer(consumerConfiguration, ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG); } KafkaBinderConfigurationProperties(KafkaProperties kafkaProperties); KafkaProperties getKafkaProperties(); Transaction getTransaction(); String getKafkaConnectionString(); String getDefaultKafkaConnectionString(); String[] getHeaders(); String[] getBrokers(); void setBrokers(String... brokers); void setDefaultBrokerPort(String defaultBrokerPort); void setHeaders(String... headers); String getRequiredAcks(); void setRequiredAcks(String requiredAcks); short getReplicationFactor(); void setReplicationFactor(short replicationFactor); int getMinPartitionCount(); void setMinPartitionCount(int minPartitionCount); int getHealthTimeout(); void setHealthTimeout(int healthTimeout); boolean isAutoCreateTopics(); void setAutoCreateTopics(boolean autoCreateTopics); boolean isAutoAddPartitions(); void setAutoAddPartitions(boolean autoAddPartitions); Map<String, String> getConfiguration(); void setConfiguration(Map<String, String> configuration); Map<String, String> getConsumerProperties(); void setConsumerProperties(Map<String, String> consumerProperties); Map<String, String> getProducerProperties(); void setProducerProperties(Map<String, String> producerProperties); Map<String, Object> mergedConsumerConfiguration(); Map<String, Object> mergedProducerConfiguration(); JaasLoginModuleConfiguration getJaas(); void setJaas(JaasLoginModuleConfiguration jaas); String getHeaderMapperBeanName(); void setHeaderMapperBeanName(String headerMapperBeanName); Duration getAuthorizationExceptionRetryInterval(); void setAuthorizationExceptionRetryInterval(Duration authorizationExceptionRetryInterval); boolean isConsiderDownWhenAnyPartitionHasNoLeader(); void setConsiderDownWhenAnyPartitionHasNoLeader(boolean considerDownWhenAnyPartitionHasNoLeader);  }
@Test public void mergedConsumerConfigurationFiltersGroupIdFromKafkaProperties() { KafkaProperties kafkaProperties = new KafkaProperties(); kafkaProperties.getConsumer().setGroupId("group1"); KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = new KafkaBinderConfigurationProperties(kafkaProperties); Map<String, Object> mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.GROUP_ID_CONFIG); }
@Test public void mergedConsumerConfigurationFiltersEnableAutoCommitFromKafkaProperties() { KafkaProperties kafkaProperties = new KafkaProperties(); kafkaProperties.getConsumer().setEnableAutoCommit(true); KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties = new KafkaBinderConfigurationProperties(kafkaProperties); Map<String, Object> mergedConsumerConfiguration = kafkaBinderConfigurationProperties.mergedConsumerConfiguration(); assertThat(mergedConsumerConfiguration).doesNotContainKeys(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); }