diff --git a/README.adoc b/README.adoc index ab71e40b13..a5e010cc21 100644 --- a/README.adoc +++ b/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 ---- 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 adminProps, + KafkaProperties bootProps, KafkaBinderConfigurationProperties binderProps) { // First deal with the outlier String kafkaConnectionString = binderProps.getKafkaConnectionString(); if (ObjectUtils diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/KafkaTopicUtils.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/KafkaTopicUtils.java index e31cbd34ba..8057245bac 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/KafkaTopicUtils.java +++ b/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 */ diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index c14ff3a9b9..373633961e 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/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. *

* Provides both producer and consumer bindings for the bound KStream. * diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index 0a0c6d49a7..d391b308a4 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/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. *

* Provides only consumer binding for the bound KTable as output bindings are not allowed * on it. diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java index 1fdd10ecd6..1d074df691 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java +++ b/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 buildDetails(KafkaStreams kafkaStreams) throws Exception { final Map details = new HashMap<>(); - final Map perAppdIdDetails = new HashMap<>(); + final Map perAppIdDetails = new HashMap<>(); boolean isRunningResult; if (isKafkaStreams25) { @@ -176,10 +176,10 @@ private Map 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); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java index 83c078d2bd..53d1d686fa 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java +++ b/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"); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index 3565dbc246..8a95e1bb64 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/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 kafkaStreamsOnlyResolvableTypes = new HashMap<>(); - private final Map kafakStreamsOnlyMethods = new HashMap<>(); + private final Map 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 { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java index 378d56214c..57a790842a 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java +++ b/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 configuration; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 9d6a95b047..9ad1d15cf4 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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( diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index 6b44a13c79..f4e17985bd 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/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. * diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java index 8bfbdd5e8c..e5f2a625af 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarConsumerProperties.java +++ b/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; diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java index a276fb27ad..74dfc6d56d 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/properties/PulsarProducerProperties.java +++ b/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; diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index edb660de9f..8388b96138 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java +++ b/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) { diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java index b43bba67e3..3463e5b63a 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java @@ -45,7 +45,7 @@ public class RabbitTestSupport private final boolean management; private final int ampqPort; - private final int managmentPort; + private final int managementPort; public RabbitTestSupport() { this(false); @@ -59,7 +59,7 @@ public RabbitTestSupport(boolean management, int amqpPort, int managementPort) { super("RABBIT"); this.management = management; this.ampqPort = amqpPort; - this.managmentPort = managementPort; + this.managementPort = managementPort; } @@ -68,7 +68,7 @@ protected void obtainResource() throws Exception { resource = new CachingConnectionFactory("localhost", this.ampqPort); resource.createConnection().close(); if (management) { - Socket socket = SocketFactory.getDefault().createSocket("localhost", this.managmentPort); + Socket socket = SocketFactory.getDefault().createSocket("localhost", this.managementPort); socket.close(); } } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java index adb1ac50d5..a11f430783 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java @@ -60,7 +60,7 @@ public class RabbitExpressionEvaluatingInterceptor implements ChannelInterceptor /** * Construct an instance with the provided expressions and evaluation context. At * least one expression muse be non-null. - * @param routingKeyExpression the routing key expresssion. + * @param routingKeyExpression the routing key expression. * @param delayExpression the delay expression. * @param evaluationContext the evaluation context. */ diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 50247b1e95..09df95a19e 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -661,7 +661,7 @@ protected PolledConsumerResources createPolledConsumerResources(String name, ExtendedConsumerProperties consumerProperties) { Assert.isTrue(!consumerProperties.isMultiplex(), - "The Spring Integration polled MessageSource does not currently support muiltiple queues"); + "The Spring Integration polled MessageSource does not currently support multiple queues"); AmqpMessageSource source = new AmqpMessageSource(this.connectionFactory, destination.getName()); source.setRawMessageHeader(true); @@ -815,9 +815,9 @@ private MessageProperties adjustMessagePropertiesHeader(Throwable cause, String messageProperties.getReceivedExchange()); headers.put(RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY, messageProperties.getReceivedRoutingKey()); - if (properties.getExtension().getRepublishDeliveyMode() != null) { + if (properties.getExtension().getRepublishDeliveryMode() != null) { messageProperties.setDeliveryMode( - properties.getExtension().getRepublishDeliveyMode()); + properties.getExtension().getRepublishDeliveryMode()); } messageProperties.incrementRetryCount(); return messageProperties; @@ -890,7 +890,7 @@ private boolean checkDlx() { * Traverse the cause tree, stopping at AmqpRejectAndDontRequeueException * or ImmediateAcknowledgeAmqpException. * @param throwable the throwable. - * @return true if neither found or AmqpRejectAndDontRequeueException is + * @return true if neither found nor AmqpRejectAndDontRequeueException is * found first. */ private boolean shouldRepublish(Throwable throwable) { diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index cdd7ae993b..93c2e9ba23 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -99,7 +99,7 @@ RabbitMessageChannelBinder rabbitMessageChannelBinder( + "#" + nameIncrementer.getAndIncrement()); } - ListenerContainerCustomizer composedCistomizer = new ListenerContainerCustomizer<>() { + ListenerContainerCustomizer composedCustomizer = new ListenerContainerCustomizer<>() { @Override public void configure(MessageListenerContainer container, String destinationName, String group) { if (!CollectionUtils.isEmpty(listenerContainerCustomizers)) { @@ -122,7 +122,7 @@ public void configure(MessageHandler handler, String destinationName) { RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder( this.rabbitConnectionFactory, this.rabbitProperties, - provisioningProvider(declarableCustomizers), composedCistomizer, sourceCustomizer); + provisioningProvider(declarableCustomizers), composedCustomizer, sourceCustomizer); binder.setAdminAddresses( this.rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 28da59a3ec..1dc9d22c28 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -1320,7 +1320,7 @@ private void testAutoBindDLQPartionedConsumerFirstWithRepublishGuts( properties.getExtension().setAutoBindDlq(true); properties.getExtension().setRepublishToDlq(true); properties.getExtension() - .setRepublishDeliveyMode(MessageDeliveryMode.NON_PERSISTENT); + .setRepublishDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); properties.setMaxAttempts(withRetry ? 2 : 1); properties.setPartitioned(true); properties.setInstanceIndex(0); diff --git a/bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java b/bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java index b9c615914d..7b41dc6a57 100644 --- a/bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java +++ b/bus/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/DefaultBusPathMatcher.java @@ -41,10 +41,10 @@ public class DefaultBusPathMatcher implements PathMatcher { private static final Log log = LogFactory.getLog(DefaultBusPathMatcher.class); - private final PathMatcher delagateMatcher; + private final PathMatcher delegateMatcher; - public DefaultBusPathMatcher(PathMatcher delagateMatcher) { - this.delagateMatcher = delagateMatcher; + public DefaultBusPathMatcher(PathMatcher delegateMatcher) { + this.delegateMatcher = delegateMatcher; } protected boolean matchMultiProfile(String pattern, String idToMatch) { @@ -83,7 +83,7 @@ protected boolean matchMultiProfile(String pattern, String idToMatch) { } for (String id : idsWithSingleProfile) { - if (this.delagateMatcher.match(pattern, id)) { + if (this.delegateMatcher.match(pattern, id)) { if (log.isDebugEnabled()) { log.debug("matched true"); } @@ -99,7 +99,7 @@ protected boolean matchMultiProfile(String pattern, String idToMatch) { @Override public boolean isPattern(String path) { - return this.delagateMatcher.isPattern(path); + return this.delegateMatcher.isPattern(path); } @Override @@ -107,7 +107,7 @@ public boolean match(String pattern, String path) { if (log.isDebugEnabled()) { log.debug("In match: " + pattern + ", " + path); } - if (!this.delagateMatcher.match(pattern, path)) { + if (!this.delegateMatcher.match(pattern, path)) { return matchMultiProfile(pattern, path); } return true; @@ -115,27 +115,27 @@ public boolean match(String pattern, String path) { @Override public boolean matchStart(String pattern, String path) { - return this.delagateMatcher.matchStart(pattern, path); + return this.delegateMatcher.matchStart(pattern, path); } @Override public String extractPathWithinPattern(String pattern, String path) { - return this.delagateMatcher.extractPathWithinPattern(pattern, path); + return this.delegateMatcher.extractPathWithinPattern(pattern, path); } @Override public Map extractUriTemplateVariables(String pattern, String path) { - return this.delagateMatcher.extractUriTemplateVariables(pattern, path); + return this.delegateMatcher.extractUriTemplateVariables(pattern, path); } @Override public Comparator getPatternComparator(String path) { - return this.delagateMatcher.getPatternComparator(path); + return this.delegateMatcher.getPatternComparator(path); } @Override public String combine(String pattern1, String pattern2) { - return this.delagateMatcher.combine(pattern1, pattern2); + return this.delegateMatcher.combine(pattern1, pattern2); } } diff --git a/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java index ad29ea2684..8ce14aa6d1 100644 --- a/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -104,7 +104,7 @@ protected Message receive(PollableChannel channel) { * Attempt to receive a message on the given channel, waiting up to 1s * * additionalMultiplier * {@link #timeoutMultiplier}). * - * Allows accomodating tests which are slower than normal (e.g. retry). + * Allows accommodating tests which are slower than normal (e.g. retry). */ protected Message receive(PollableChannel channel, int additionalMultiplier) { long startTime = System.currentTimeMillis(); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 495c38f1fc..78d7f61cd6 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -477,7 +477,7 @@ protected void postProcessOutputChannel(MessageChannel outputChannel, * the binder must wire this channel into the producer endpoint so that errors are * forwarded to it. * @return the message handler for sending data to the target middleware - * @throws Exception when producer messsage handler failed to be created + * @throws Exception when producer message handler failed to be created */ protected MessageHandler createProducerMessageHandler(ProducerDestination destination, P producerProperties, MessageChannel channel, MessageChannel errorChannel) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index 5399504674..854f3439cc 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -83,7 +83,7 @@ default void stop() { } /** - * Will always return false unless overriden. + * Will always return false unless overridden. */ @Override default boolean isPaused() { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java index 24f73e690a..cbaa509ca5 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java @@ -19,7 +19,7 @@ import org.springframework.context.ApplicationEvent; /** - * ApplicationEvent fired whenever the the Binding is created. + * ApplicationEvent fired whenever the Binding is created. * * @author Oleg Zhurakousky * @since 2.0 diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index c6f3c9bb41..d971bbd2cb 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java @@ -126,7 +126,7 @@ public class ConsumerProperties { /** * Allows you to further qualify which RetryTemplate to use for a specific consumer - * binding.. + * binding. */ private String retryTemplateName; diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index af53992139..c3f69c154d 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -358,7 +358,7 @@ private void populateCandidatesForBindableType(Class bindingTar * Return true if the binder is a {@link PollableConsumerBinder} and the target type * is a {@link PollableSource} and their generic types match (e.g. MessageHandler), OR * if it's a {@link Binder} and the target matches the binder's generic type. - * @param bindng target type + * @param binding target type * @param binderInstance the binder. * @param bindingTargetType the binding target type. * @return true if the conditions match. @@ -385,7 +385,7 @@ private Binder getBinderInstance( if (binderConfiguration != null) { this.flatten(null, binderConfiguration.getProperties(), binderProperties); } - binderProducingContext = this.createUnitializedContextForAOT(configurationName, binderProperties, binderConfiguration); + binderProducingContext = this.createInitializedContextForAOT(configurationName, binderProperties, binderConfiguration); this.binderChildContextInitializers.get(configurationName).initialize(binderProducingContext); registerOuterContextBean(binderProperties, binderProducingContext); binderProducingContext.refresh(); @@ -611,7 +611,7 @@ private Class loadClass(String className) { * @param binderConfiguration binder configuration * @return a binder child application context suitable for AOT initialization */ - GenericApplicationContext createUnitializedContextForAOT(String configurationName, + GenericApplicationContext createInitializedContextForAOT(String configurationName, Map binderProperties, BinderConfiguration binderConfiguration) { GenericApplicationContext binderContext = new GenericApplicationContext(); // Set the conversion service on the binder producing context to handle complex properties diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/EmbeddedHeaderUtils.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/EmbeddedHeaderUtils.java index 00aff8f1c6..5cdbcccfd5 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/EmbeddedHeaderUtils.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/EmbeddedHeaderUtils.java @@ -61,7 +61,7 @@ public static String decodeExceptionMessage(Message requestMessage) { * been embedded into the new message payload. * * @param original original message - * @param headers headers to embedd + * @param headers headers to embedded * @return a new message */ public static byte[] embedHeaders(MessageValues original, String... headers) { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExpressionSerializer.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExpressionSerializer.java index 38b654b4da..417b71d422 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExpressionSerializer.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExpressionSerializer.java @@ -25,8 +25,8 @@ import tools.jackson.databind.ser.std.StdSerializer; /** - * since 5.0.2 - * author Oleg Zhurakousky + * @author Oleg Zhurakousky + * @since 5.0.2 */ public class ExpressionSerializer extends StdSerializer { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java index 3662f77ed5..3ead16fb40 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java @@ -22,7 +22,7 @@ /** * A MessageHandler that always is the last subscriber (on a {@link BinderErrorChannel}) - * that throws an exception if it the only subscriber (aside from the bridge to the global + * that throws an exception if the only subscriber (aside from the bridge to the global * error channel). It is typically only used if a binder implementation does not return a * handled from {@code getErrorMessageHandler()}. * diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageValues.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageValues.java index 6401f09e6c..99882a4c71 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageValues.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageValues.java @@ -87,7 +87,7 @@ public Message toMessage(MessageBuilderFactory messageBuilderFactory) { } /** - * Convert to a {@link Message} using a the default + * Convert to a {@link Message} using the default * {@link org.springframework.integration.support.MessageBuilder}. * @return the Message */ diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java index ed93046157..d8e45825e7 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java @@ -57,10 +57,9 @@ public void configureOutputChannel(MessageChannel messageChannel, String channel @Override public void configurePolledMessageSource(PollableMessageSource binding, String name) { - this.messageChannelConfigurers.forEach(cconfigurer -> { - if (cconfigurer instanceof MessageChannelAndSourceConfigurer) { - ((MessageChannelAndSourceConfigurer) cconfigurer) - .configurePolledMessageSource(binding, name); + this.messageChannelConfigurers.forEach(configure -> { + if (configure instanceof MessageChannelAndSourceConfigurer messageChannelAndSourceConfigurer) { + messageChannelAndSourceConfigurer.configurePolledMessageSource(binding, name); } }); } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index 6016a7ca33..e4bda976c5 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -109,7 +109,7 @@ public class BindingServiceProperties private Map bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); /** - * Additional per-binder properties (see {@link BinderProperties}) if more then one + * Additional per-binder properties (see {@link BinderProperties}) if more than one * binder of the same type is used (i.e., connect to multiple instances of RabbitMq). * Here you can specify multiple binder configurations, each with different * environment settings. For example; spring.cloud.stream.binders.rabbit1.environment. diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java index 8d7b69111e..26c48e526f 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java @@ -144,7 +144,7 @@ public MessageConverter getMessageConverterForType(MimeType mimeType) { } else { if (this.log.isDebugEnabled()) { - this.log.debug("Ommitted " + converter + " of type " + this.log.debug("Omitted " + converter + " of type " + converter.getClass().toString() + " for '" + mimeType.toString() + "' as it is not an AbstractMessageConverter"); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java index a4475bac57..1de124d0be 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java @@ -101,10 +101,9 @@ public static Class getParameterType(Class evaluatedClass, /** * Return the generic type of PollableSource to determine if it is appropriate for the - * binder. e.g., with PollableMessageSource extends - * PollableSource<MessageHandler> and AbstractMessageChannelBinder implements - * PollableConsumerBinder<MessageHandler, C> We're checking that the the generic - * type (MessageHandler) matches. + * binder. e.g., with {@code PollableMessageSource extends PollableSource} + * and {@code AbstractMessageChannelBinder implements PollableConsumerBinder}. + * We're checking that the generic type (MessageHandler) matches. * @param binderInstance the binder. * @param bindingTargetType the binding target type. * @return true if found, false otherwise. diff --git a/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc b/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc index c200cae669..cc9ce0c01c 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-binder/manual-ack.adoc @@ -5,13 +5,13 @@ This example illustrates how one may manually acknowledge offsets in a consumer This example requires that `spring.cloud.stream.kafka.bindings.input.consumer.ackMode` be set to `MANUAL`. Use the corresponding input channel name for your example. -[source] +[source,java] ---- @SpringBootApplication -public class ManuallyAcknowdledgingConsumer { +public class ManuallyAcknowledgingConsumer { public static void main(String[] args) { - SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args); + SpringApplication.run(ManuallyAcknowledgingConsumer.class, args); } @Bean diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc index d9e062374b..c6eea9866e 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/configuration-options.adoc @@ -206,7 +206,7 @@ Default: `event_type` consumedAs:: Custom name for the source component from which the processor is consuming from. + -Deafult: `none` (generated by Kafka Streams) +Default: `none` (generated by Kafka Streams) [[special-note-on-concurrency]] == Special note on concurrency diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc index cd6d2411f6..db2ec8c036 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/multi-binders-with-based-binders-and-regular-binder.adoc @@ -61,12 +61,12 @@ spring.cloud.stream.bindings.kstreamProcess-out-0.binder=kafka3 Pay attention to the above configuration. We have two kinds of binders, but 3 binders all in all, first one is the regular Kafka binder based on cluster 1 (`kafka1`), then another Kafka binder based on cluster 2 (`kafka2`) and finally the `kstream` one (`kafka3`). -The first processor in the application receives data from `kafka1` and publishes to `kafka2` where both binders are based on regular Kafka binder but differnt clusters. +The first processor in the application receives data from `kafka1` and publishes to `kafka2` where both binders are based on regular Kafka binder but different clusters. The second processor, which is a Kafka Streams processor consumes data from `kafka3` which is the same cluster as `kafka2`, but a different binder type. Since there are three different binder types available in the Kafka Streams family of binders - `kstream`, `ktable` and `globalktable` - if your application has multiple bindings based on any of these binders, that needs to be explicitly provided as the binder type. -For e.g if you have a processor as below, +For e.g. if you have a processor as below, ``` @Bean diff --git a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc index 4935c0de37..5136b511a7 100644 --- a/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka-streams-binder/programming-model.adoc @@ -119,7 +119,7 @@ If you want to multiplex multiple topics into a single `KStream` binding, you ca `spring.cloud.stream.bindings.process-in-0.destination=topic-1,topic-2,topic-3` -In addition, you can also provide topic patterns as destinations if you want to match topics against a regular exression. +In addition, you can also provide topic patterns as destinations if you want to match topics against a regular expression. `spring.cloud.stream.bindings.process-in-0.destination=input.*` diff --git a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc index e943b9db4a..67a3ddf9d0 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_overview.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_overview.adoc @@ -524,10 +524,10 @@ Use the corresponding input channel name for your example. [source,java] ---- @SpringBootApplication -public class ManuallyAcknowdledgingConsumer { +public class ManuallyAcknowledgingConsumer { public static void main(String[] args) { - SpringApplication.run(ManuallyAcknowdledgingConsumer.class, args); + SpringApplication.run(ManuallyAcknowledgingConsumer.class, args); } @Bean diff --git a/docs/modules/ROOT/pages/kafka/kafka_tips.adoc b/docs/modules/ROOT/pages/kafka/kafka_tips.adoc index 9be650d70f..f32ed35d1f 100644 --- a/docs/modules/ROOT/pages/kafka/kafka_tips.adoc +++ b/docs/modules/ROOT/pages/kafka/kafka_tips.adoc @@ -384,7 +384,7 @@ Let's say that the new consumer gets partitions `2` and `3`. When this new Spring Cloud Stream consumer calls this `onPartitionsAssigned` method, it will see that this is the initial assignment for partition `2` and `3` on this consumer. Therefore, it will do the seek operation because of the conditional check on the `initial` argument. In the case of the first consumer, it now only has partitions `0` and `1` -However, for this consumer it was simply a rebalance event and not considered as an intial assignment. +However, for this consumer it was simply a rebalance event and not considered as an initial assignment. Thus, it will not re-seek to the given offsets because of the conditional check on the `initial` argument. [[how-do-i-manually-acknowledge-using-kafka-binder?]] @@ -503,7 +503,7 @@ I need to send a key along with the payload of the record, is there a way to do It is often necessary that you want to send associative data structure like a map as the record with a key and value. Spring Cloud Stream allows you to do that in a straightforward manner. -Following is a basic blueprint for doing this, but you may want to adapt it to your paricular use case. +Following is a basic blueprint for doing this, but you may want to adapt it to your particular use case. Here is sample producer method (aka `Supplier`). diff --git a/docs/modules/ROOT/pages/preface.adoc b/docs/modules/ROOT/pages/preface.adoc index f1bfac50f5..3abbf6e4ff 100644 --- a/docs/modules/ROOT/pages/preface.adoc +++ b/docs/modules/ROOT/pages/preface.adoc @@ -44,7 +44,7 @@ When the "`Cloud Stream`" option should appears, select it. + Basically, you choose the messaging middleware to which your application binds. We recommend using the one you have already installed or feel more comfortable with installing and running. -Also, as you can see from the Initilaizer screen, there are a few other options you can choose. +Also, as you can see from the Initializr screen, there are a few other options you can choose. For example, you can choose Gradle as your build tool instead of Maven (the default). . In the *Artifact* field, type 'logging-consumer'. + diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_overview/prod-props.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_overview/prod-props.adoc index cbc883839a..14da050392 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_overview/prod-props.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_overview/prod-props.adoc @@ -9,7 +9,7 @@ in the format of `spring.cloud.stream.rabbit.default.producer.= Also, keep in mind that binding specific property will override its equivalent in the default. -altermateExchange.binding.queue:: +alternateExchange.binding.queue:: If the exchange does not already exist, and a `name` is provided, bind this queue to the alternate exchange. A simple durable queue with no arguments is provisioned; if more sophisticated configuration is required, you must configure and bind the queue yourself. + diff --git a/docs/modules/ROOT/pages/rabbit/rabbit_partitions.adoc b/docs/modules/ROOT/pages/rabbit/rabbit_partitions.adoc index 6fc75ac6a2..5c9a8e9bec 100644 --- a/docs/modules/ROOT/pages/rabbit/rabbit_partitions.adoc +++ b/docs/modules/ROOT/pages/rabbit/rabbit_partitions.adoc @@ -63,7 +63,7 @@ public class RabbitPartitionProducerApplication { [NOTE] ==== -The configuration in the prececing example uses the default partitioning (`key.hashCode() % partitionCount`). +The configuration in the preceding example uses the default partitioning (`key.hashCode() % partitionCount`). This may or may not provide a suitably balanced algorithm, depending on the key values. You can override this default by using the `partitionSelectorExpression` or `partitionSelectorClass` properties. diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/binding-names.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/binding-names.adoc index 202b7531dc..ebdff1df98 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/binding-names.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/binding-names.adoc @@ -9,5 +9,5 @@ Throughout this manual you will see examples of configuration properties such as The `input` segment in this property name is what we refer to as _binding name_ and it could derive via several mechanisms. The following sub-sections will describe the naming conventions and configuration elements used by spring-cloud-stream to control binding names. -NOTE: If your binding name has special characters, such as the `.` character, you need to surround the binding key with brackets (`[]`) and then wrap it in qoutes. +NOTE: If your binding name has special characters, such as the `.` character, you need to surround the binding key with brackets (`[]`) and then wrap it in quotes. For example `spring.cloud.stream.bindings."[my.output.binding.key]".destination`. diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc index 5c243d23b3..ee0533f0bc 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/binding_visualization_control.adoc @@ -50,7 +50,7 @@ bindingProperties.setConsumer(consumerProperties); RabbitConsumerProperties extendedConsumerProperties = controller.createInputBinding(inputName, "rabbit", bindingProperties); ---- -You can than manage its properties by calling `getExtensionProperties(..)` method. +You can then manage its properties by calling `getExtensionProperties(..)` method. [source,java] ---- diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/event-routing.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/event-routing.adoc index 04bd09e149..2513b5f4e0 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/event-routing.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/event-routing.adoc @@ -186,7 +186,7 @@ public class SourceWithDynamicDestination { ---- Albeit trivial you can clearly see in this example, our output is a Message with `spring.cloud.stream.sendto.destination` header -set to the value of he input argument. The framework will consult this header and will attempt to create or discover +set to the value of the input argument. The framework will consult this header and will attempt to create or discover a destination with that name and send output to it. diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/explicit-binding-creation.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/explicit-binding-creation.adoc index 573f1ed325..d15d217c81 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/explicit-binding-creation.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/explicit-binding-creation.adoc @@ -41,7 +41,7 @@ do not require additional suffix such as `in-0/out-0` given that: - the binding name is not backed by a function - it already represents _input_ or _output_ binding by definition - - it can not have more then one input or output. + - it can not have more than one input or output. So `spring.cloud.stream.output-bindings=hello` will result in the binding named `hello` (not `hello-out-0`) and you can further refer to it as `spring.cloud.stream.bindings.hello.destination=foobar` \ No newline at end of file diff --git a/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc b/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc index 43dab453b0..8685e45275 100644 --- a/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc +++ b/docs/modules/ROOT/pages/spring-cloud-stream/overview-error-handling.adoc @@ -22,7 +22,7 @@ That is because anything you read later in this section with regard to Retry Tem DLQ and configuration properties that assist with all of it ***only*** applies to Message handlers (i.e., imperative functions). Reactive API provides a very rich library of its own operators and mechanisms to assist you with error handling specific to -variety of reactive uses cases which are far more complex then simple Message handler cases, So use them, such +variety of reactive uses cases which are far more complex than simple Message handler cases, So use them, such as `public final Flux retryWhen(Retry retrySpec);` that you can find in `reactor.core.publisher.Flux`. [source,java] diff --git a/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AvroMessageConverterProperties.java b/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AvroMessageConverterProperties.java index f23f94d739..4b0cc4118b 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AvroMessageConverterProperties.java +++ b/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AvroMessageConverterProperties.java @@ -34,9 +34,8 @@ public class AvroMessageConverterProperties { /** * The source directory of Apache Avro schema. This schema is used by this converter. - * If this schema depends on other schemas consider defining those those dependent - * ones in the {@link #schemaImports} - * @parameter + * If this schema depends on other schemas consider defining those dependent + * ones in the {@link #schemaImports}. */ private Resource[] schemaLocations; @@ -44,7 +43,6 @@ public class AvroMessageConverterProperties { * A list of files or directories that should be loaded first thus making them * importable by subsequent schemas. Note that imported files should not reference * each other. - * @parameter */ private Resource[] schemaImports; diff --git a/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/EnableSchemaRegistryServer.java b/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/EnableSchemaRegistryServer.java index b0f9f32dd9..b8504d3e51 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/EnableSchemaRegistryServer.java +++ b/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/EnableSchemaRegistryServer.java @@ -26,7 +26,7 @@ import org.springframework.context.annotation.Import; /** - * Enables the schema registry server enpoints. + * Enables the schema registry server endpoints. * * @author Vinicius Carvalho */ diff --git a/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/model/Compatibility.java b/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/model/Compatibility.java index f62f21f1b4..7f3cdcec73 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/model/Compatibility.java +++ b/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/model/Compatibility.java @@ -22,7 +22,7 @@ public enum Compatibility { /** - * Backward compatibiltity. + * Backward compatibility. */ BACKWARD, diff --git a/tools/kafka/docker-compose/README.adoc b/tools/kafka/docker-compose/README.adoc index 9d96255bc9..09899feacc 100644 --- a/tools/kafka/docker-compose/README.adoc +++ b/tools/kafka/docker-compose/README.adoc @@ -74,7 +74,7 @@ docker-compose -f ./kafka-cluster.yml -f ./schema-registry.yml down [#_all_the_things] === All The Things -The scripts can be chained together to start the cluster, UI and schema registry with the following commmand: +The scripts can be chained together to start the cluster, UI and schema registry with the following command: [source,shell] ---- docker-compose -f kafka-cluster.yml -f ./control-center-ui.yml -f ./schema-registry.yml up