Fix typos and remove incorrect tags

README.adoc

@@ -282,7 +282,7 @@ Checkstyle rules are *disabled by default*. To add checkstyle to your project ju
 
 If you need to suppress some rules (e.g. line length needs to be longer), then it's enough for you to define a file under `${project.root}/src/checkstyle/checkstyle-suppressions.xml` with your suppressions. Example:
 
-.projectRoot/src/checkstyle/checkstyle-suppresions.xml
+.projectRoot/src/checkstyle/checkstyle-suppressions.xml
 ----
 <?xml version="1.0"?>
 <!DOCTYPE suppressions PUBLIC
@@ -359,7 +359,7 @@ Spring Cloud Build brings along the  `basepom:duplicate-finder-maven-plugin`, th
 
 ==== Duplicate Finder configuration
 
-Duplicate finder is *enabled by default* and will run in the `verify` phase of your Maven build, but it will only take effect in your project if you add the `duplicate-finder-maven-plugin` to the `build` section of the projecst's `pom.xml`.
+Duplicate finder is *enabled by default* and will run in the `verify` phase of your Maven build, but it will only take effect in your project if you add the `duplicate-finder-maven-plugin` to the `build` section of the project's `pom.xml`.
 
 .pom.xml
 [source,xml]

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/common/BinderHeaderMapper.java

@@ -439,7 +439,7 @@ private class MimeTypeJsonDeserializer extends StdNodeBasedDeserializer<MimeType
 
 		@Override
 		public MimeType convert(JsonNode root, DeserializationContext ctxt) {
-			if (root instanceof StringNode	) {
+			if (root instanceof StringNode) {
 				return MimeType.valueOf(root.asText());
 			}
 			else {

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java

@@ -131,7 +131,7 @@
 
 	/**
 	 * Time between retries after AuthorizationException is caught in
-	 * the ListenerContainer; defalt is null which disables retries.
+	 * the ListenerContainer; default is null which disables retries.
 	 * For more info see: {@link org.springframework.kafka.listener.ConsumerProperties#setAuthorizationExceptionRetryInterval(java.time.Duration)}
 	 */
 	private Duration authorizationExceptionRetryInterval;
@@ -400,7 +400,7 @@
 	 * @return the merged properties.
 	 */
 	public Map<String, Object> mergedConsumerConfiguration() {
 		Map<String, Object> consumerConfiguration = new HashMap<>(this.kafkaProperties.buildConsumerProperties());
 		if (!consumerConfiguration.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG) && this.kafkaConnectionDetails != null) {
 			consumerConfiguration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaConnectionDetails.getConsumer().getBootstrapServers());
 		}
@@ -429,7 +429,7 @@
 	 * @return the merged properties.
 	 */
 	public Map<String, Object> mergedProducerConfiguration() {
 		Map<String, Object> producerConfiguration = new HashMap<>(this.kafkaProperties.buildProducerProperties());
 		if (!producerConfiguration.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG) && this.kafkaConnectionDetails != null) {
 			producerConfiguration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaConnectionDetails.getProducer().getBootstrapServers());
 		}

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -205,7 +205,7 @@ public enum StandardHeaders {
 	private String transactionManager;
 
 	/**
-	 * Set to false to NOT commit the offset of a successfully recovered recovered in the after rollback processor.
+	 * Set to false to NOT commit the offset of a successfully recovered in the after rollback processor.
 	 */
 	private boolean txCommitRecovered = true;
 
@@ -493,7 +493,7 @@ public void setPollTimeout(long pollTimeout) {
 	/**
 	 * @return the transaction manager bean name.
 	 *
-	 * Transaction manager bean name (must be {@code KafkaAwareTransactionManager}.
+	 * Transaction manager bean name (must be {@code KafkaAwareTransactionManager}).
 	 */
 	public String getTransactionManager() {
 		return this.transactionManager;

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java

@@ -275,7 +275,7 @@ public void setRecordMetadataChannel(String recordMetadataChannel) {
 	/**
 	 * @return the transaction manager bean name.
 	 *
-	 * Transaction manager bean name (must be {@code KafkaAwareTransactionManager}.
+	 * Transaction manager bean name (must be {@code KafkaAwareTransactionManager}).
 	 */
 	public String getTransactionManager() {
 		return this.transactionManager;

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -134,7 +134,7 @@ public KafkaTopicProvisioner(
 	 * Create an instance.
 	 * @param kafkaBinderConfigurationProperties the binder configuration properties.
 	 * @param kafkaProperties the boot Kafka properties used to build the instance.
-	 * @parak kafkaConnectionDetails the Kafka connection details used to build the instance
+	 * @param kafkaConnectionDetails the Kafka connection details used to build the instance
 	 * @param adminClientConfigCustomizer to customize {@link AdminClient}.
 	 * @since 4.1.4
 	 */
@@ -166,7 +166,7 @@ public KafkaTopicProvisioner(
 	 *
 	 * @param kafkaBinderConfigurationProperties the binder configuration properties.
 	 * @param kafkaProperties the boot Kafka properties used to build the instance.
-	 * @param kafkaConnectionDetails the Kafka connection deatils used to build the instance.
+	 * @param kafkaConnectionDetails the Kafka connection details used to build the instance.
 	 * @param adminClientConfigCustomizers to customize {@link AdminClient}.
 	 * @since 4.1.4
 	 */
@@ -178,7 +178,7 @@ public KafkaTopicProvisioner(
 		Assert.isTrue(kafkaProperties != null, "KafkaProperties cannot be null");
 		this.configurationProperties = kafkaBinderConfigurationProperties;
 		this.adminClientProperties = createAdminClientProperties(kafkaProperties, kafkaConnectionDetails);
-		normalalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties,
+		normalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties,
 			kafkaBinderConfigurationProperties);
 		// If the application provides AdminConfig customizers
 		// and overrides properties, those take precedence.
@@ -337,9 +337,25 @@ AdminClient createAdminClient() {
 	 * @param adminProps the admin properties to normalize.
 	 * @param bootProps the boot kafka properties.
 	 * @param binderProps the binder kafka properties.
+	 * @deprecated use {@link #normalizeBootPropsWithBinder(Map, KafkaProperties, KafkaBinderConfigurationProperties)} instead.
 	 */
+	@Deprecated(forRemoval = true)
 	public static void normalalizeBootPropsWithBinder(Map<String, Object> adminProps,
 			KafkaProperties bootProps, KafkaBinderConfigurationProperties binderProps) {
+		normalizeBootPropsWithBinder(adminProps, bootProps, binderProps);
+	}
+
+	/**
+	 * In general, binder properties supersede boot kafka properties. The one exception is
+	 * the bootstrap servers. In that case, we should only override the boot properties if
+	 * (there is a binder property AND it is a non-default value) OR (if there is no boot
+	 * property); this is needed because the binder property never returns a null value.
+	 * @param adminProps the admin properties to normalize.
+	 * @param bootProps the boot kafka properties.
+	 * @param binderProps the binder kafka properties.
+	 */
+	public static void normalizeBootPropsWithBinder(Map<String, Object> adminProps,
+			KafkaProperties bootProps, KafkaBinderConfigurationProperties binderProps) {
 		// First deal with the outlier
 		String kafkaConnectionString = binderProps.getKafkaConnectionString();
 		if (ObjectUtils

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/KafkaTopicUtils.java

@@ -19,7 +19,7 @@
 import java.nio.charset.StandardCharsets;
 
 /**
- * Utility methods releated to Kafka topics.
+ * Utility methods related to Kafka topics.
  *
  * @author Soby Chacko
  */

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java

@@ -49,7 +49,7 @@
 
 /**
  * {@link org.springframework.cloud.stream.binder.Binder} implementation for
- * {@link KStream}. This implemenation extends from the {@link AbstractBinder} directly.
+ * {@link KStream}. This implementation extends from the {@link AbstractBinder} directly.
  * <p>
  * Provides both producer and consumer bindings for the bound KStream.
  *

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java

@@ -41,7 +41,7 @@
 
 /**
  * {@link org.springframework.cloud.stream.binder.Binder} implementation for
- * {@link KTable}. This implemenation extends from the {@link AbstractBinder} directly.
+ * {@link KTable}. This implementation extends from the {@link AbstractBinder} directly.
  * <p>
  * Provides only consumer binding for the bound KTable as output bindings are not allowed
  * on it.

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java

@@ -96,7 +96,7 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator i
 		kafkaProperties.buildAdminProperties();
 		this.configurationProperties = kafkaStreamsBinderConfigurationProperties;
 		this.adminClientProperties = kafkaProperties.buildAdminProperties();
-		KafkaTopicProvisioner.normalalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties,
+		KafkaTopicProvisioner.normalizeBootPropsWithBinder(this.adminClientProperties, kafkaProperties,
 				kafkaStreamsBinderConfigurationProperties);
 		this.kafkaStreamsRegistry = kafkaStreamsRegistry;
 		this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue;
@@ -150,7 +150,7 @@ protected void doHealthCheck(Health.Builder builder) throws Exception {
 
 	private Map<String, Object> buildDetails(KafkaStreams kafkaStreams) throws Exception {
 		final Map<String, Object> details = new HashMap<>();
-		final Map<String, Object> perAppdIdDetails = new HashMap<>();
+		final Map<String, Object> perAppIdDetails = new HashMap<>();
 
 		boolean isRunningResult;
 		if (isKafkaStreams25) {
@@ -176,10 +176,10 @@ private Map<String, Object> buildDetails(KafkaStreams kafkaStreams) throws Excep
 				threadDetail.put("standbyTasks", taskDetails(metadata.standbyTasks()));
 				threadDetails.put(metadata.threadName(), threadDetail);
 			}
-			perAppdIdDetails.put("threadDetails", threadDetails);
+			perAppIdDetails.put("threadDetails", threadDetails);
 			final StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.kafkaStreamsRegistry.streamBuilderFactoryBean(kafkaStreams);
 			final String applicationId = (String) streamsBuilderFactoryBean.getStreamsConfiguration().get(StreamsConfig.APPLICATION_ID_CONFIG);
-			details.put(applicationId, perAppdIdDetails);
+			details.put(applicationId, perAppIdDetails);
 		}
 		else {
 			final StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.kafkaStreamsRegistry.streamBuilderFactoryBean(kafkaStreams);

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java

@@ -78,7 +78,7 @@ public class KeyValueSerdeResolver implements ApplicationContextAware {
 	 * Provide the {@link Serde} for inbound key.
 	 * @param extendedConsumerProperties binding level extended
 	 * {@link KafkaStreamsConsumerProperties}
-	 * @return configurd {@link Serde} for the inbound key.
+	 * @return configured {@link Serde} for the inbound key.
 	 */
 	public Serde<?> getInboundKeySerde(
 			KafkaStreamsConsumerProperties extendedConsumerProperties) {
@@ -99,7 +99,7 @@ public Serde<?> getInboundKeySerde(
 	 * @param consumerProperties {@link ConsumerProperties} on binding
 	 * @param extendedConsumerProperties binding level extended
 	 * {@link KafkaStreamsConsumerProperties}
-	 * @return configurd {@link Serde} for the inbound value.
+	 * @return configured {@link Serde} for the inbound value.
 	 */
 	public Serde<?> getInboundValueSerde(ConsumerProperties consumerProperties,
 			KafkaStreamsConsumerProperties extendedConsumerProperties) {
@@ -143,7 +143,7 @@ public Serde<?> getInboundValueSerde(ConsumerProperties consumerProperties,
 	/**
 	 * Provide the {@link Serde} for outbound key.
 	 * @param properties binding level extended {@link KafkaStreamsProducerProperties}
-	 * @return configurd {@link Serde} for the outbound key.
+	 * @return configured {@link Serde} for the outbound key.
 	 */
 	public Serde<?> getOuboundKeySerde(KafkaStreamsProducerProperties properties) {
 		return getKeySerde(properties.getKeySerde(), properties.getConfiguration());
@@ -159,7 +159,7 @@ public Serde<?> getOuboundKeySerde(KafkaStreamsProducerProperties properties, Re
 	 * @param producerProperties {@link ProducerProperties} on binding
 	 * @param kafkaStreamsProducerProperties binding level extended
 	 * {@link KafkaStreamsProducerProperties}
-	 * @return configurd {@link Serde} for the outbound value.
+	 * @return configured {@link Serde} for the outbound value.
 	 */
 	public Serde<?> getOutboundValueSerde(ProducerProperties producerProperties,
 			KafkaStreamsProducerProperties kafkaStreamsProducerProperties) {
@@ -246,7 +246,7 @@ private Serde<?> getKeySerde(String keySerdeString, ResolvableType resolvableTyp
 			else {
 				keySerde = Serdes.ByteArray();
 				if (resolvableType != null &&
-						(isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
+						(isResolvableKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) {
 					ResolvableType targetType = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0);
 					Serde<?> fallbackSerde = getFallbackSerde("default.key.serde");
 					keySerde = SerdeResolverUtils.resolveForType(this.context, targetType, fallbackSerde);
@@ -265,7 +265,7 @@ private boolean isResolvableKStreamArrayType(ResolvableType resolvableType) {
 				KStream.class.isAssignableFrom(Objects.requireNonNull(resolvableType.getComponentType().getRawClass()));
 	}
 
-	private boolean isResolvalbeKafkaStreamsType(ResolvableType resolvableType) {
+	private boolean isResolvableKafkaStreamsType(ResolvableType resolvableType) {
 		return resolvableType.getRawClass() != null && (KStream.class.isAssignableFrom(resolvableType.getRawClass()) || KTable.class.isAssignableFrom(resolvableType.getRawClass()) ||
 				GlobalKTable.class.isAssignableFrom(resolvableType.getRawClass()));
 	}
@@ -300,7 +300,7 @@ private Serde<?> getValueSerde(String valueSerdeString, ResolvableType resolvabl
 		}
 		else {
 			valueSerde = Serdes.ByteArray();
-			if (resolvableType != null && ((isResolvalbeKafkaStreamsType(resolvableType)) ||
+			if (resolvableType != null && ((isResolvableKafkaStreamsType(resolvableType)) ||
 					(isResolvableKStreamArrayType(resolvableType)))) {
 				ResolvableType targetType = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1);
 				Serde<?> fallbackSerde = getFallbackSerde("default.value.serde");

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java

@@ -80,7 +80,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean,
 	private final StreamFunctionProperties streamFunctionProperties;
 
 	private final Map<String, ResolvableType> kafkaStreamsOnlyResolvableTypes = new HashMap<>();
-	private final Map<String, Method> kafakStreamsOnlyMethods = new HashMap<>();
+	private final Map<String, Method> kafkaStreamsOnlyMethods = new HashMap<>();
 	private ConfigurableApplicationContext applicationContext;
 
 	public KafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties streamFunctionProperties) {
@@ -114,7 +114,7 @@ public void afterPropertiesSet() {
 				.forEach(this::extractResolvableTypes);
 
 		kafkaStreamsOnlyResolvableTypes.keySet().forEach(k -> addResolvableTypeInfo(k, kafkaStreamsOnlyResolvableTypes.get(k)));
-		kafakStreamsOnlyMethods.keySet().forEach(k -> addResolvableTypeInfo(k, kafakStreamsOnlyMethods.get(k)));
+		kafkaStreamsOnlyMethods.keySet().forEach(k -> addResolvableTypeInfo(k, kafkaStreamsOnlyMethods.get(k)));
 
 		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
 
@@ -250,7 +250,7 @@ private void saveMethodInfoForComponentBeans(String key, Method method) {
 			this.methods.put(key, method);
 		}
 		else {
-			kafakStreamsOnlyMethods.put(key, method);
+			kafkaStreamsOnlyMethods.put(key, method);
 		}
 	}
 
@@ -283,7 +283,7 @@ private void discoverOnlyKafkaStreamsResolvableTypes(String key, ResolvableType
 	}
 
 	private void addResolvableTypeInfo(String key, Method method) {
-		if (kafakStreamsOnlyMethods.size() == 1) {
+		if (kafkaStreamsOnlyMethods.size() == 1) {
 			this.methods.put(key, method);
 		}
 		else {

binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java

@@ -124,7 +124,7 @@ public static class Functions {
 		private String applicationId;
 
 		/**
-		 * Funcion specific configuraiton to use.
+		 * Function specific configuration to use.
 		 */
 		private Map<String, String> configuration;
 

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -1016,7 +1016,7 @@ public void onPartitionsAssigned(Consumer<?, ?> consumer,
 	}
 
 	/*
-	 * Reset the offsets if needed; may update the offsets in in the container's
+	 * Reset the offsets if needed; may update the offsets in the container's
 	 * topicPartitionInitialOffsets.
 	 */
 	private void resetOffsetsForAutoRebalance(

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java

@@ -93,7 +93,7 @@
 public class KafkaBinderConfiguration {
 
 	/**
-	 * @ConfigurationProperties is declared on the @Bean method for Spring Boot to ignore
+	 * {@code @ConfigurationProperties} is declared on the @Bean method for Spring Boot to ignore
 	 * constructor binding on KafkaBinderConfigurationProperties. If constructor binding is
 	 * used, it ignores all the JavaBeans style properties when generating configuration metadata.
 	 *

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java

@@ -42,7 +42,7 @@ public class PulsarConsumerProperties extends ConsumerConfigProperties {
 
 	/**
 	 * Pulsar message key type for this binding (only used when schema type is
-	 * {@code }KEY_VALUE}).
+	 * {@code KEY_VALUE}).
 	 */
 	@Nullable
 	private Class<?> messageKeyType;

binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java

@@ -42,7 +42,7 @@ public class PulsarProducerProperties extends ProducerConfigProperties {
 
 	/**
 	 * Pulsar message key type for this binding (only used when schema type is
-	 * {@code }KEY_VALUE}).
+	 * {@code KEY_VALUE}).
 	 */
 	@Nullable
 	private Class<?> messageKeyType;

binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java

@@ -39,7 +39,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
 
 	/**
-	 * maxumum concurrency of this consumer (threads).
+	 * maximum concurrency of this consumer (threads).
 	 */
 	private int maxConcurrency = 1;
 
@@ -66,7 +66,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	/**
 	 * when republishing to the DLQ, the delivery mode to use.
 	 */
-	private MessageDeliveryMode republishDeliveyMode = MessageDeliveryMode.PERSISTENT;
+	private MessageDeliveryMode republishDeliveryMode = MessageDeliveryMode.PERSISTENT;
 
 	/**
 	 * true to requeue rejected messages, false to discard (or route to DLQ).
@@ -222,12 +222,36 @@ public boolean isRequeueRejected() {
 		return requeueRejected;
 	}
 
+	/**
+	 * @return the message delivery mode used when republishing.
+	 * @deprecated use {@link #getRepublishDeliveryMode()} instead.
+	 */
+	@Deprecated(forRemoval = true)
 	public MessageDeliveryMode getRepublishDeliveyMode() {
-		return this.republishDeliveyMode;
+		return this.republishDeliveryMode;
+	}
+
+	/**
+	 * @param republishDeliveryMode the message delivery mode to use when republishing.
+	 * @deprecated use {@link #setRepublishDeliveryMode(MessageDeliveryMode)} instead.
+	 */
+	@Deprecated(forRemoval = true)
+	public void setRepublishDeliveyMode(MessageDeliveryMode republishDeliveryMode) {
+		this.republishDeliveryMode = republishDeliveryMode;
 	}
 
-	public void setRepublishDeliveyMode(MessageDeliveryMode republishDeliveyMode) {
-		this.republishDeliveyMode = republishDeliveyMode;
+	/**
+	 * @return the message delivery mode used when republishing.
+	 */
+	public MessageDeliveryMode getRepublishDeliveryMode() {
+		return this.republishDeliveryMode;
+	}
+
+	/**
+	 * @param republishDeliveryMode the message delivery mode to use when republishing.
+	 */
+	public void setRepublishDeliveryMode(MessageDeliveryMode republishDeliveryMode) {
+		this.republishDeliveryMode = republishDeliveryMode;
 	}
 
 	public void setRequeueRejected(boolean requeueRejected) {