Fix cross references

This commit is contained in:
Marcin Grzejszczak
2023-09-08 16:12:38 +02:00
committed by Oleg Zhurakousky
parent f2cff0c609
commit bb4186b856
27 changed files with 76 additions and 76 deletions

View File

@@ -9,16 +9,16 @@ Sabby Anandan; Marius Bogoevici; Eric Bottard; Mark Fisher; Ilayaperumal Gopinat
The reference documentation consists of the following sections:
[horizontal]
<<spring-cloud-stream.adoc#spring-cloud-stream-reference,Overview>> :: History, Quick Start, Concepts, Architecture Overview, Binder Abstraction, and Core Features
xref:spring-cloud-stream.adoc#spring-cloud-stream-reference[Overview] :: History, Quick Start, Concepts, Architecture Overview, Binder Abstraction, and Core Features
<<spring-cloud-stream-binder-rabbit#spring-cloud-stream-binder-rabbit-reference,Rabbit MQ Binder>> :: Spring Cloud Stream binder reference for Rabbit MQ
xref:rabbit/spring-cloud-stream-binder-rabbit.adoc[Rabbit MQ Binder] :: Spring Cloud Stream binder reference for Rabbit MQ
<<spring-cloud-stream-binder-kafka#_apache_kafka_binder,Apache Kafka Binder>> :: Spring Cloud Stream binder reference for Apache Kafka
<<spring-cloud-stream-binder-kafka#_kafka_streams_binder,Apache Kafka Streams binder>> :: Spring Cloud Stream binder reference for Apache Kafka Streams
<<spring-cloud-stream-binder-pulsar#spring-cloud-stream-binder-pulsar-reference,Apache Pulsar Binder>> :: Spring Cloud Stream binder reference for Apache Pulsar
xref:pulsar/spring-cloud-stream-binder-pulsar.adoc[Apache Pulsar Binder] :: Spring Cloud Stream binder reference for Apache Pulsar
https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#spring-cloud-stream-binder-for-solace-pubsub[Solace PubSub+ Binder] :: Spring Cloud Stream binder reference for Solace PubSub+ (Partner Maintained)
<<binders.adoc#binders,Additional Binders>> :: A collection of Partner maintained binder implementations for Spring Cloud Stream (e.g., Azure Event Hubs, Google PubSub, Solace PubSub+)
<<spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-schema-registry-reference,Schema Registry>> :: Spring Cloud Stream Schema Registry Reference
xref:spring-cloud-stream/overview-error-handling.adoc#spring-cloud-stream-overview-binders[Additional Binders] :: A collection of Partner maintained binder implementations for Spring Cloud Stream (e.g., Azure Event Hubs, Google PubSub, Solace PubSub+)
xref:schema-registry/spring-cloud-stream-schema-registry.adoc[Schema Registry] :: Spring Cloud Stream Schema Registry Reference
https://github.com/spring-cloud/spring-cloud-stream-samples/[Spring Cloud Stream Samples] :: A curated collection of repeatable Spring Cloud Stream samples to walk through the features
Relevant Links:

View File

@@ -88,7 +88,7 @@ When transactions are enabled, individual `producer` properties are ignored and
Default `null` (no transactions)
spring.cloud.stream.kafka.binder.transaction.producer.*::
Global producer properties for producers in a transactional binder.
See `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` and <<kafka-producer-properties>> and the general producer properties supported by all binders.
See `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` and xref:kafka/kafka_overview/configuration-options.adoc#kafka-producer-properties[Kafka Producer Properties] and the general producer properties supported by all binders.
+
Default: See individual producer properties.
@@ -195,15 +195,15 @@ Note that this property is only applicable for pollable consumers.
Default: not set.
resetOffsets::
Whether to reset offsets on the consumer to the value provided by startOffset.
Must be false if a `KafkaBindingRebalanceListener` is provided; see <<rebalance-listener>>.
See <<reset-offsets>> for more information about this property.
Must be false if a `KafkaBindingRebalanceListener` is provided; see xref:kafka/kafka_overview/rebalance-listener.adoc[Using a KafkaBindingRebalanceListener].
See xref:kafka/kafka_overview/configuration-options.adoc#reset-offsets[Resetting Offsets] for more information about this property.
+
Default: `false`.
startOffset::
The starting offset for new groups.
Allowed values: `earliest` and `latest`.
If the consumer group is set explicitly for the consumer 'binding' (through `spring.cloud.stream.bindings.<channelName>.group`), 'startOffset' is set to `earliest`. Otherwise, it is set to `latest` for the `anonymous` consumer group.
See <<reset-offsets>> for more information about this property.
See xref:kafka/kafka_overview/configuration-options.adoc#reset-offsets[Resetting Offsets] for more information about this property.
+
Default: null (equivalent to `earliest`).
enableDlq::
@@ -211,17 +211,17 @@ When set to true, it enables DLQ behavior for the consumer.
By default, messages that result in errors are forwarded to a topic named `error.<destination>.<group>`.
The DLQ topic name can be configurable by setting the `dlqName` property or by defining a `@Bean` of type `DlqDestinationResolver`.
This provides an alternative option to the more common Kafka replay scenario for the case when the number of errors is relatively small and replaying the entire original topic may be too cumbersome.
See <<kafka-dlq-processing>> processing for more information.
See xref:kafka/kafka_dlq.adoc[Dead-Letter Topic Processing] processing for more information.
Starting with version 2.0, messages sent to the DLQ topic are enhanced with the following headers: `x-original-topic`, `x-exception-message`, and `x-exception-stacktrace` as `byte[]`.
By default, a failed record is sent to the same partition number in the DLQ topic as the original record.
See <<dlq-partition-selection>> for how to change that behavior.
See xref:kafka/kafka_dlq.adoc#dlq-partition-selection[Dead-Letter Topic Partition Selection] for how to change that behavior.
**Not allowed when `destinationIsPattern` is `true`.**
+
Default: `false`.
dlqPartitions::
When `enableDlq` is true, and this property is not set, a dead letter topic with the same number of partitions as the primary topic(s) is created.
Usually, dead-letter records are sent to the same partition in the dead-letter topic as the original record.
This behavior can be changed; see <<dlq-partition-selection>>.
This behavior can be changed; see xref:kafka/kafka_dlq.adoc#dlq-partition-selection[Dead-Letter Topic Partition Selection].
If this property is set to `1` and there is no `DqlPartitionFunction` bean, all dead-letter records will be written to partition `0`.
If this property is greater than `1`, you **MUST** provide a `DlqPartitionFunction` bean.
Note that the actual partition count is affected by the binder's `minPartitionCount` property.
@@ -258,7 +258,7 @@ Default: `null`
idleEventInterval::
The interval, in milliseconds, between events indicating that no messages have recently been received.
Use an `ApplicationListener<ListenerContainerIdleEvent>` to receive these events.
See <<pause-resume>> for a usage example.
See xref:kafka/kafka_overview/configuration-options.adoc#pause-resume[Example: Pausing and Resuming the Consumer] for a usage example.
+
Default: `30000`
destinationIsPattern::
@@ -331,7 +331,7 @@ Set `resetOffsets` to `true` and `startOffset` to `latest`; the binding will per
IMPORTANT: If a rebalance occurs after the initial assignment, the seeks will only be performed on any newly assigned partitions that were not assigned during the initial assignment.
For more control over topic offsets, see <<rebalance-listener>>; when a listener is provided, `resetOffsets` should not be set to `true`, otherwise, that will cause an error.
For more control over topic offsets, see xref:kafka/kafka_overview/rebalance-listener.adoc[Using a KafkaBindingRebalanceListener]; when a listener is provided, `resetOffsets` should not be set to `true`, otherwise, that will cause an error.
[[consuming-batches]]
== Consuming Batches
@@ -437,7 +437,7 @@ The header contains a `RecordMetadata` object provided by the Kafka client; it i
+
`ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)`
+
Failed sends go the producer error channel (if configured); see <<kafka-error-channels>>.
Failed sends go the producer error channel (if configured); see xref:kafka/kafka_overview/error-channels.adoc[Error Channels].
+
Default: null.

View File

@@ -9,6 +9,6 @@ The payload of the `ErrorMessage` for a send failure is a `KafkaSendFailureExcep
* `failedMessage`: The Spring Messaging `Message<?>` that failed to be sent.
* `record`: The raw `ProducerRecord` that was created from the `failedMessage`
There is no automatic handling of producer exceptions (such as sending to a <<kafka-dlq-processing, Dead-Letter queue>>).
There is no automatic handling of producer exceptions (such as sending to a xref:kafka/kafka_dlq.adoc[Dead-Letter queue]).
You can consume these exceptions with your own Spring Integration flow.

View File

@@ -10,7 +10,7 @@ The metric contains the consumer group information, topic and the actual lag in
This metric is particularly useful for providing auto-scaling feedback to a PaaS platform.
The metric collection behaviour can be configured by setting properties in the `spring.cloud.stream.kafka.binder.metrics` namespace,
refer to the <<kafka-binder-properties, kafka binder properties section>> for more information.
refer to the xref:kafka/kafka_overview/configuration-options.adoc#kafka-binder-properties[kafka binder properties section] for more information.
You can exclude `KafkaBinderMetrics` from creating the necessary infrastructure like consumers and then reporting the metrics by providing the following component in the application.

View File

@@ -80,7 +80,7 @@ ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate<?, ?> template) {
Now, only a single retry delay needs to be greater than the consumer's `max.poll.interval.ms` property.
When working with several binders, the 'ListenerContainerWithDlqAndRetryCustomizer' bean gets overridden by the 'DefaultBinderFactory'. For the bean
to apply, you need to use a 'BinderCustomizer' to set the container customizer (See <<binder-customizer>>):
to apply, you need to use a 'BinderCustomizer' to set the container customizer (See xref:spring-cloud-stream/binder-customizer.adoc[Customizing binders in multi binder applications]):
====
[source, java]
----

View File

@@ -3,7 +3,7 @@
This section contains the configuration options used by the Kafka Streams binder.
For common configuration options and properties pertaining to binder, refer to the <<binding-properties,core documentation>>.
For common configuration options and properties pertaining to binder, refer to the xref:spring-cloud-stream/binding-properties.adoc[core documentation].
[[kafka-streams-binder-properties]]
== Kafka Streams Binder Properties

View File

@@ -29,9 +29,9 @@ and abstractions, making it easier to digest the rest of this user guide.
The three steps are as follows:
. <<spring-cloud-stream-preface-creating-sample-application>>
. <<spring-cloud-stream-preface-importing-project>>
. <<spring-cloud-stream-preface-adding-message-handler>>
. xref:preface.adoc#spring-cloud-stream-preface-creating-sample-application[Creating a Sample Application by Using Spring Initializr]
. xref:preface.adoc#spring-cloud-stream-preface-importing-project[Importing the Project into Your IDE]
. xref:preface.adoc#spring-cloud-stream-preface-adding-message-handler[Adding a Message Handler, Building, and Running]
[[spring-cloud-stream-preface-creating-sample-application]]
== Creating a Sample Application by Using Spring Initializr
@@ -124,7 +124,7 @@ As you can see from the preceding listing:
Doing so also lets you see one of the core features of the framework: It tries to automatically convert incoming message payloads to type `Person`.
You now have a fully functional Spring Cloud Stream application that does listens for messages.
From here, for simplicity, we assume you selected RabbitMQ in <<spring-cloud-stream-preface-creating-sample-application,step one>>.
From here, for simplicity, we assume you selected RabbitMQ in xref:preface.adoc#spring-cloud-stream-preface-creating-sample-application[step one].
Assuming you have RabbitMQ installed and running, you can start the application by running its `main` method in your IDE.
You should see following output:

View File

@@ -47,11 +47,11 @@ If retry is enabled (`maxAttempts > 1`), failed messages are delivered to the DL
If retry is disabled (`maxAttempts = 1`), you should set `requeueRejected` to `false` (the default) so that failed messages are routed to the DLQ, instead of being re-queued.
In addition, `republishToDlq` causes the binder to publish a failed message to the DLQ (instead of rejecting it).
This feature lets additional information (such as the stack trace in the `x-exception-stacktrace` header) be added to the message in headers.
See the <<spring-cloud-stream-rabbit-frame-max-headroom, `frameMaxHeadroom` property>> for information about truncated stack traces.
See the xref:rabbit/rabbit_overview/rabbitmq-consumer-properties.adoc#spring-cloud-stream-rabbit-frame-max-headroom[`frameMaxHeadroom` property] for information about truncated stack traces.
This option does not need retry enabled.
You can republish a failed message after just one attempt.
Starting with version 1.2, you can configure the delivery mode of republished messages.
See the <<spring-cloud-stream-rabbit-republish-delivery-mode,`republishDeliveryMode` property>>.
See the xref:rabbit/rabbit_overview/rabbitmq-consumer-properties.adoc#spring-cloud-stream-rabbit-republish-delivery-mode[`republishDeliveryMode` property].
If the stream listener throws an `ImmediateAcknowledgeAmqpException`, the DLQ is bypassed and the message simply discarded.
Starting with version 2.1, this is true regardless of the setting of `republishToDlq`; previously it was only the case when `republishToDlq` was `false`.
@@ -64,10 +64,10 @@ This allows the transaction to roll back if the publishing fails for some reason
In addition, if the connection factory is configured for publisher confirms or returns, the publication to the DLQ will wait for the confirmation and check for a returned message.
If a negative acknowledgment or returned message is received, the binder will throw an `AmqpRejectAndDontRequeueException`, allowing the broker to take care of publishing to the DLQ as if the `republishToDlq` property is `false`.
See <<rabbit-binder-properties>> for more information about these properties.
See xref:rabbit/rabbit_overview/binder-properties.adoc[RabbitMQ Binder Properties] for more information about these properties.
The framework does not provide any standard mechanism to consume dead-letter messages (or to re-route them back to the primary queue).
Some options are described in <<rabbit-dlq-processing>>.
Some options are described in xref:rabbit/rabbit_dlq.adoc[Dead-Letter Queue Processing].
NOTE: When multiple RabbitMQ binders are used in a Spring Cloud Stream application, it is important to disable 'RabbitAutoConfiguration' to avoid the same configuration from `RabbitAutoConfiguration` being applied to the two binders.
You can exclude the class by using the `@SpringBootApplication` annotation.

View File

@@ -38,7 +38,7 @@ batchingEnabled::
Whether to enable message batching by producers.
Messages are batched into one message according to the following properties (described in the next three entries in this list): 'batchSize', `batchBufferLimit`, and `batchTimeout`.
See https://docs.spring.io/spring-amqp//reference/html/_reference.html#template-batching[Batching] for more information.
Also see <<rabbit-receiving-batch>>.
Also see xref:rabbit/rabbit_overview/receiving-batch.adoc[Receiving Batched Messages].
+
Default: `false`.
batchSize::
@@ -304,7 +304,7 @@ Applies only when `requiredGroups` are provided and then only to those groups.
+
Default: `no limit`
useConfirmHeader::
See <<publisher-confirms>>.
See xref:rabbit/rabbit_overview/publisher-confirms.adoc[Publisher Confirms].
Mutually exclusive with `confirmAckChannel`.
+

View File

@@ -2,7 +2,7 @@
= Publisher Confirms
There are two mechanisms to get the result of publishing a message; in each case, the connection factory must have `publisherConfirmType` set `ConfirmType.CORRELATED`.
The "legacy" mechanism is to set the `confirmAckChannel` to the bean name of a message channel from which you can retrieve the confirmations asynchronously; negative acks are sent to the error channel (if enabled) - see <<rabbit-error-channels>>.
The "legacy" mechanism is to set the `confirmAckChannel` to the bean name of a message channel from which you can retrieve the confirmations asynchronously; negative acks are sent to the error channel (if enabled) - see xref:rabbit/rabbit_overview/putting-it-all-together.adoc#rabbit-error-channels[Error Channels].
The preferred mechanism, added in version 3.1 is to use a correlation data header and wait for the result via its `Future<Confirm>` property.
This is particularly useful with a batch listener because you can send multiple messages before waiting for the result.

View File

@@ -50,7 +50,7 @@ Notice that the count property in the `x-death` header is a `Long`.
= Error Channels
Starting with version 1.3, the binder unconditionally sends exceptions to an error channel for each consumer destination and can also be configured to send async producer send failures to an error channel.
See "`<<spring-cloud-stream-overview-error-handling>>`" for more information.
See "`xref:spring-cloud-stream/overview-error-handling.adoc[Error Handling]`" for more information.
RabbitMQ has two types of send failures:
@@ -61,7 +61,7 @@ The latter is rare.
According to the RabbitMQ documentation "[A nack] will only be delivered if an internal error occurs in the Erlang process responsible for a queue.".
You can also get a negative acknowledgment if you publish to a bounded queue with `reject-publish` queue overflow behavior.
As well as enabling producer error channels (as described in "`<<spring-cloud-stream-overview-error-handling>>`"), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows:
As well as enabling producer error channels (as described in "`xref:spring-cloud-stream/overview-error-handling.adoc[Error Handling]`"), the RabbitMQ binder only sends messages to the channels if the connection factory is appropriately configured, as follows:
* `ccf.setPublisherConfirms(true);`
* `ccf.setPublisherReturns(true);`
@@ -80,14 +80,14 @@ The payload of the `ErrorMessage` for a returned message is a `ReturnedAmqpMessa
* `exchange`: The exchange to which the message was published.
* `routingKey`: The routing key used when the message was published.
Also see <<publisher-confirms>> for an alternative mechanism to receive returned messages.
Also see xref:rabbit/rabbit_overview/publisher-confirms.adoc[Publisher Confirms] for an alternative mechanism to receive returned messages.
For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageException` with the following properties:
* `failedMessage`: The spring-messaging `Message<?>` that failed to be sent.
* `nackReason`: A reason (if available -- you may need to examine the broker logs for more information).
There is no automatic handling of these exceptions (such as sending to a <<rabbit-dlq-processing, dead-letter queue>>).
There is no automatic handling of these exceptions (such as sending to a xref:rabbit/rabbit_dlq.adoc[dead-letter queue]).
You can consume these exceptions with your own Spring Integration flow.
[[rabbit-binder-health-indicator]]

View File

@@ -275,7 +275,7 @@ republishToDlq::
By default, messages that fail after retries are exhausted are rejected.
If a dead-letter queue (DLQ) is configured, RabbitMQ routes the failed message (unchanged) to the DLQ.
If set to `true`, the binder republishs failed messages to the DLQ with additional headers, including the exception message and stack trace from the cause of the final failure.
Also see the <<spring-cloud-stream-rabbit-frame-max-headroom, frameMaxHeadroom property>>.
Also see the xref:rabbit/rabbit_overview/rabbitmq-consumer-properties.adoc#spring-cloud-stream-rabbit-frame-max-headroom[frameMaxHeadroom property].
+
Default: `true`
singleActiveConsumer::

View File

@@ -102,7 +102,7 @@ There are similar properties used when declaring a dead-letter exchange/queue, w
When retry is enabled within the binder, the listener container thread is suspended for any back off periods that are configured.
This might be important when strict ordering is required with a single consumer. However, for other use cases, it prevents other messages from being processed on that thread.
An alternative to using binder retry is to set up dead lettering with time to live on the dead-letter queue (DLQ) as well as dead-letter configuration on the DLQ itself.
See "`<<rabbit-binder-properties>>`" for more information about the properties discussed here.
See "`xref:rabbit/rabbit_overview/binder-properties.adoc[RabbitMQ Binder Properties]`" for more information about the properties discussed here.
You can use the following example configuration to enable this feature:
* Set `autoBindDlq` to `true`.

View File

@@ -6,7 +6,7 @@ With the RabbitMQ binder, there are two types of batches handled by consumer bin
[[batches-created-by-producers]]
== Batches Created by Producers
Normally, if a producer binding has `batch-enabled=true` (see <<rabbit-prod-props>>), or a message is created by a `BatchingRabbitTemplate`, elements of the batch are returned as individual calls to the listener method.
Normally, if a producer binding has `batch-enabled=true` (see xref:rabbit/rabbit_overview/prod-props.adoc[Rabbit Producer Properties]), or a message is created by a `BatchingRabbitTemplate`, elements of the batch are returned as individual calls to the listener method.
Starting with version 3.0, any such batch can be presented as a `List<?>` to the listener method if `spring.cloud.stream.bindings.<name>.consumer.batch-mode` is set to `true`.
[[consumer-side-batching]]

View File

@@ -238,13 +238,13 @@ You can customize the schema storage by using the http://docs.spring.io/spring-b
The Schema Registry Server API consists of the following operations:
* `POST /` -- see `<<spring-cloud-stream-overview-registering-new-schema>>`
* `GET /{subject}/{format}/{version}` -- see `<<spring-cloud-stream-overview-retrieve-schema-subject-format-version>>`
* `GET /{subject}/{format}` -- see `<<spring-cloud-stream-overview-retrieve-schema-subject-format>>`
* `GET /schemas/{id}` -- see `<<spring-cloud-stream-overview-retrieve-schema-id>>`
* `DELETE /{subject}/{format}/{version}` -- see `<<spring-cloud-stream-overview-deleting-schema-subject-format-version>>`
* `DELETE /schemas/{id}` -- see `<<spring-cloud-stream-overview-deleting-schema-id>>`
* `DELETE /{subject}` -- see `<<spring-cloud-stream-overview-deleting-schema-subject>>`
* `POST /` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-registering-new-schema[Registering a New Schema]`
* `GET /{subject}/{format}/{version}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-subject-format-version[Retrieving an Existing Schema by Subject, Format, and Version]`
* `GET /{subject}/{format}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-subject-format[Retrieving an Existing Schema by Subject and Format]`
* `GET /schemas/{id}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-retrieve-schema-id[Retrieving an Existing Schema by ID]`
* `DELETE /{subject}/{format}/{version}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-subject-format-version[Deleting a Schema by Subject, Format, and Version]`
* `DELETE /schemas/{id}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-id[Deleting a Schema by ID]`
* `DELETE /{subject}` -- see `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-deleting-schema-subject[Deleting a Schema by Subject]`
[[spring-cloud-stream-overview-registering-new-schema]]
===== Registering a New Schema
@@ -347,8 +347,8 @@ NOTE: The ConfluentSchemaRegistryClient is tested against Confluent platform ver
To better understand how Spring Cloud Stream registers and resolves new schemas and its use of Avro schema comparison features, we provide two separate subsections:
* `<<spring-cloud-stream-overview-schema-registration-process>>`
* `<<spring-cloud-stream-overview-schema-resolution-process>>`
* `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-schema-registration-process[Schema Registration Process (Serialization)]`
* `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-schema-resolution-process[Schema Resolution Process (Deserialization)]`
[[spring-cloud-stream-overview-schema-registration-process]]
==== Schema Registration Process (Serialization)
@@ -372,7 +372,7 @@ With the schema version information, the converter sets the `contentType` header
[[spring-cloud-stream-overview-schema-resolution-process]]
==== Schema Resolution Process (Deserialization)
When reading messages that contain version information (that is, a `contentType` header with a scheme like the one described under `<<spring-cloud-stream-overview-schema-registration-process>>`, the converter queries the Schema server to fetch the writer schema of the message.
When reading messages that contain version information (that is, a `contentType` header with a scheme like the one described under `xref:schema-registry/spring-cloud-stream-schema-registry.adoc#spring-cloud-stream-overview-schema-registration-process[Schema Registration Process (Serialization)]`, the converter queries the Schema server to fetch the writer schema of the message.
Once it has found the correct schema of the incoming message, it retrieves the reader schema and, by using Avro's schema resolution support, reads it into the reader definition (setting defaults and any missing properties).
.Schema Reading Resolution Process

View File

@@ -77,10 +77,10 @@ class BootTestStreamApplicationTests {
Spring Cloud Stream provides a number of abstractions and primitives that simplify the writing of message-driven microservice applications.
This section gives an overview of the following:
* <<spring-cloud-stream-overview-application-model,Spring Cloud Stream's application model>>
* <<spring-cloud-stream-overview-binder-abstraction>>
* <<spring-cloud-stream-overview-persistent-publish-subscribe-support,Persistent publish-subscribe support>>
* <<consumer-groups,Consumer group support>>
* <<partitioning,Partitioning support>>
* <<spring-cloud-stream-overview-binder-api,A pluggable Binder SPI>>
* xref:spring-cloud-stream/overview-application-model.adoc[Spring Cloud Stream's application model]
* xref:spring-cloud-stream/overview-binder-abstraction.adoc[The Binder Abstraction]
* xref:spring-cloud-stream/overview-persistent-publish-subscribe-support.adoc[Persistent publish-subscribe support]
* xref:spring-cloud-stream/consumer-groups.adoc[Consumer group support]
* xref:spring-cloud-stream/overview-partitioning.adoc[Partitioning support]
* xref:spring-cloud-stream/overview-binder-api.adoc[A pluggable Binder SPI]

View File

@@ -37,17 +37,17 @@ The default value of this property cannot be overridden.
group::
The consumer group of the binding.
Applies only to inbound bindings.
See <<consumer-groups,Consumer Groups>>.
See xref:spring-cloud-stream/consumer-groups.adoc[Consumer Groups].
+
Default: `null` (indicating an anonymous consumer).
contentType::
The content type of this binding.
See `<<content-type-management>>`.
See `xref:spring-cloud-stream/binding-properties.adoc#content-type-management[Content Type Negotiation]`.
+
Default: `application/json`.
binder::
The binder used by this binding.
See `<<multiple-binders>>` for details.
See `xref:spring-cloud-stream/multiple-binders.adoc[Multiple Binders on the Classpath]` for details.
+
Default: `null` (the default binder is used, if it exists).
@@ -104,14 +104,14 @@ Default: `true`.
instanceCount::
When set to a value greater than equal to zero, it allows customizing the instance count of this consumer (if different from `spring.cloud.stream.instanceCount`).
When set to a negative value, it defaults to `spring.cloud.stream.instanceCount`.
See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
See `xref:spring-cloud-stream/overview-instance-index-instance-count.adoc[Instance Index and Instance Count]` for more information.
+
Default: `-1`.
instanceIndex::
When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`).
When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`.
Ignored if `instanceIndexList` is provided.
See `<<spring-cloud-stream-overview-instance-index-instance-count>>` for more information.
See `xref:spring-cloud-stream/overview-instance-index-instance-count.adoc[Instance Index and Instance Count]` for more information.
+
Default: `-1`.
instanceIndexList::
@@ -179,7 +179,7 @@ Default: `true`.
partitionKeyExpression::
A SpEL expression that determines how to partition outbound data.
If set, outbound data on this binding is partitioned. `partitionCount` must be set to a value greater than 1 to be effective.
See `<<partitioning>>`.
See `xref:spring-cloud-stream/overview-partitioning.adoc[Partitioning]`.
+
Default: null.

View File

@@ -22,7 +22,7 @@ Default: empty (letting any destination be bound).
spring.cloud.stream.defaultBinder::
The default binder to use, if multiple binders are configured.
See <<multiple-binders,Multiple Binders on the Classpath>>.
See xref:spring-cloud-stream/multiple-binders.adoc[Multiple Binders on the Classpath].
+
Default: empty.
@@ -30,7 +30,7 @@ spring.cloud.stream.overrideCloudConnectors::
This property is only applicable when the `cloud` profile is active and Spring Cloud Connectors are provided with the application.
If the property is `false` (the default), the binder detects a suitable bound service (for example, a RabbitMQ service bound in Cloud Foundry for the RabbitMQ binder) and uses it for creating connections (usually through Spring Cloud Connectors).
When set to `true`, this property instructs binders to completely ignore the bound services and rely on Spring Boot properties (for example, relying on the `spring.rabbitmq.*` properties provided in the environment for the RabbitMQ binder).
The typical usage of this property is to be nested in a customized environment <<multiple-systems, when connecting to multiple systems>>.
The typical usage of this property is to be nested in a customized environment xref:spring-cloud-stream/multiple-systems.adoc[when connecting to multiple systems].
+
Default: `false`.

View File

@@ -22,4 +22,4 @@ The following is the list of available binder implementations
As it was mentioned earlier Binder abstraction is also one of the extension points of the framework. So if you can't find a suitable binder in the preceding list you can implement your own binder on top of Spring Cloud Stream.
In the https://medium.com/@domenicosibilio/how-to-create-a-spring-cloud-stream-binder-from-scratch-ab8b29ee931b[How to create a Spring Cloud Stream Binder from scratch] post a community member documents
in details, with an example, a set of steps necessary to implement a custom binder.
The steps are also highlighted in the `<<spring-cloud-stream-overview-custom-binder-impl>>` section.
The steps are also highlighted in the `xref:spring-cloud-stream/overview-custom-binder-impl.adoc[Implementing Custom Binders]` section.

View File

@@ -56,7 +56,7 @@ operation of the MessageConverter takes `targetClass` as one of its arguments.
The framework also ensures that the provided `Message` always contains a `contentType` header.
When no contentType header was already present, it injects either the per-binding `contentType` header or the default `contentType` header.
The combination of `contentType` argument type is the mechanism by which framework determines if message can be converted to a target type.
If no appropriate `MessageConverter` is found, an exception is thrown, which you can handle by adding a custom `MessageConverter` (see `<<spring-cloud-stream-overview-user-defined-message-converters>>`).
If no appropriate `MessageConverter` is found, an exception is thrown, which you can handle by adding a custom `MessageConverter` (see `xref:spring-cloud-stream/overview-user-defined-message-converters.adoc[User-defined Message Converters]`).
But what if the payload type matches the target type declared by the handler method? In this case, there is nothing to convert, and the
payload is passed unmodified. While this sounds pretty straightforward and logical, keep in mind handler methods that take a `Message<?>` or `Object` as an argument.

View File

@@ -7,13 +7,13 @@ The framework also includes a test binder for integration testing of your applic
Binder abstraction is also one of the extension points of the framework, which means you can implement your own binder on top of Spring Cloud Stream.
In the https://medium.com/@domenicosibilio/how-to-create-a-spring-cloud-stream-binder-from-scratch-ab8b29ee931b[How to create a Spring Cloud Stream Binder from scratch] post a community member documents
in details, with an example, a set of steps necessary to implement a custom binder.
The steps are also highlighted in the `<<spring-cloud-stream-overview-custom-binder-impl>>` section.
The steps are also highlighted in the `xref:spring-cloud-stream/overview-custom-binder-impl.adoc[Implementing Custom Binders]` section.
Spring Cloud Stream uses Spring Boot for configuration, and the Binder abstraction makes it possible for a Spring Cloud Stream application to be flexible in how it connects to middleware.
For example, deployers can dynamically choose, at runtime, the mapping between the external destinations (such as the Kafka topics or RabbitMQ exchanges) and inputs
and outputs of the message handler (such as input parameter of the function and its return argument).
Such configuration can be provided through external configuration properties and in any form supported by Spring Boot (including application arguments, environment variables, and `application.yml` or `application.properties` files).
In the sink example from the <<spring-cloud-stream-overview-introducing>> section, setting the `spring.cloud.stream.bindings.input.destination` application property to `raw-sensor-data` causes it to read from the `raw-sensor-data` Kafka topic or from a queue bound to the `raw-sensor-data` RabbitMQ exchange.
In the sink example from the xref:spring-cloud-stream.adoc#spring-cloud-stream-overview-introducing[Introducing Spring Cloud Stream] section, setting the `spring.cloud.stream.bindings.input.destination` application property to `raw-sensor-data` causes it to read from the `raw-sensor-data` Kafka topic or from a queue bound to the `raw-sensor-data` RabbitMQ exchange.
Spring Cloud Stream automatically detects and uses a binder found on the classpath.
You can use different types of middleware with the same code.

View File

@@ -34,5 +34,5 @@ org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration
NOTE: As it was mentioned earlier Binder abstraction is also one of the extension points of the framework. So if you can't find a suitable binder in the preceding list you can implement your own binder on top of Spring Cloud Stream.
In the https://medium.com/@domenicosibilio/how-to-create-a-spring-cloud-stream-binder-from-scratch-ab8b29ee931b[How to create a Spring Cloud Stream Binder from scratch] post a community member documents
in details, with an example, a set of steps necessary to implement a custom binder.
The steps are also highlighted in the `<<spring-cloud-stream-overview-custom-binder-impl>>` section.
The steps are also highlighted in the `xref:spring-cloud-stream/overview-custom-binder-impl.adoc[Implementing Custom Binders]` section.

View File

@@ -3,8 +3,8 @@
Partitioning in Spring Cloud Stream consists of two tasks:
* `<<spring-cloud-stream-overview-configuring-output-bindings-partitioning>>`
* `<<spring-cloud-stream-overview-configuring-input-bindings-partitioning>>`
* `xref:spring-cloud-stream/overview-partitioning.adoc#spring-cloud-stream-overview-configuring-output-bindings-partitioning[Configuring Output Bindings for Partitioning]`
* `xref:spring-cloud-stream/overview-partitioning.adoc#spring-cloud-stream-overview-configuring-input-bindings-partitioning[Configuring Input Bindings for Partitioning]`
[[spring-cloud-stream-overview-configuring-output-bindings-partitioning]]
== Configuring Output Bindings for Partitioning

View File

@@ -47,7 +47,7 @@ public class MyCustomMessageConverter extends AbstractMessageConverter {
Spring Cloud Stream enables communication between applications. Inter-application communication is a complex issue spanning several concerns, as described in the following topics:
* `<<spring-cloud-stream-overview-connecting-multiple-application-instances>>`
* `<<spring-cloud-stream-overview-instance-index-instance-count>>`
* `<<spring-cloud-stream-overview-partitioning>>`
* `xref:spring-cloud-stream/overview-connecting-multiple-application-instances.adoc[Connecting Multiple Application Instances]`
* `xref:spring-cloud-stream/overview-instance-index-instance-count.adoc[Instance Index and Instance Count]`
* `xref:spring-cloud-stream/overview-partitioning.adoc[Partitioning]`

View File

@@ -912,10 +912,10 @@ The `PollableMessageSource.poll()` method takes a `MessageHandler` argument (oft
It returns `true` if the message was received and successfully processed.
As with message-driven consumers, if the `MessageHandler` throws an exception, messages are published to error channels,
as discussed in `<<spring-cloud-stream-overview-error-handling>>`.
as discussed in `xref:spring-cloud-stream/overview-error-handling.adoc[Error Handling]`.
Normally, the `poll()` method acknowledges the message when the `MessageHandler` exits.
If the method exits abnormally, the message is rejected (not re-queued), but see <<polled-errors>>.
If the method exits abnormally, the message is rejected (not re-queued), but see xref:spring-cloud-stream/producing-and-consuming-messages.adoc#polled-errors[Handling Errors].
You can override that behavior by taking responsibility for the acknowledgment, as shown in the following example:
[source,java]

View File

@@ -11,5 +11,5 @@ It invokes Objects `toString()` method or, if the payload is `byte[]`, a new
When no appropriate converter is found, the framework throws an exception. When that happens, you should check your code and configuration and ensure you did not miss anything (that is, ensure that you provided a `contentType` by using a binding or a header).
However, most likely, you found some uncommon case (such as a custom `contentType` perhaps) and the current stack of provided `MessageConverters`
does not know how to convert. If that is the case, you can add custom `MessageConverter`. See <<spring-cloud-stream-overview-user-defined-message-converters>>.
does not know how to convert. If that is the case, you can add custom `MessageConverter`. See xref:spring-cloud-stream/overview-user-defined-message-converters.adoc[User-defined Message Converters].

View File

@@ -270,7 +270,7 @@ After testing the application with the test binder, if you want to continue doin
Spring Cloud Stream provides a health indicator for binders.
It is registered under the name `binders` and can be enabled or disabled by setting the `management.health.binders.enabled` property.
To enable health check you first need to enable both "web" and "actuator" by including its dependencies (see <<binding_visualization_control>>)
To enable health check you first need to enable both "web" and "actuator" by including its dependencies (see xref:spring-cloud-stream/binding_visualization_control.adoc[Binding visualization and control])
If `management.health.binders.enabled` is not set explicitly by the application, then `management.health.defaults.enabled` is matched as `true` and the binder health indicators are enabled.
If you want to disable health indicator completely, then you have to set `management.health.binders.enabled` to `false`.
@@ -288,7 +288,7 @@ Even if you are not disabling the binder health indicators, you can still enhanc
When you have multiple binders in the same application, health indicators are enabled by default unless the application turns them off by setting `management.health.binders.enabled` to `false`.
In this case, if the user wants to disable health check for a subset of the binders, then that should be done by setting `management.health.binders.enabled` to `false` in the multi binder configurations's environment.
See <<multiple-systems,Connecting to Multiple Systems>> for details on how environment specific properties can be provided.
See xref:spring-cloud-stream/multiple-systems.adoc[Connecting to Multiple Systems] for details on how environment specific properties can be provided.
If there are multiple binders present in the classpath but not all of them are used in the application, this may cause some issues in the context of health indicators.
There may be implementation specific details as to how the health checks are performed. For example, a Kafka binder may decide the status as `DOWN` if there are no destinations registered by the binder.