diff --git a/README.adoc b/README.adoc index fe27642c9..e385d26e7 100644 --- a/README.adoc +++ b/README.adoc @@ -162,6 +162,9 @@ Default: none. [[kafka-consumer-properties]] ==== Kafka Consumer Properties +NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.default.=`. + + The following properties are available for Kafka consumers only and must be prefixed with `spring.cloud.stream.kafka.bindings..consumer.`. @@ -284,6 +287,9 @@ Default: none (the binder-wide default of 1 is used). [[kafka-producer-properties]] ==== Kafka Producer Properties +NOTE: To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of `spring.cloud.stream.default.=`. + + The following properties are available for Kafka producers only and must be prefixed with `spring.cloud.stream.kafka.bindings..producer.`. @@ -311,7 +317,8 @@ How long the producer waits to allow more messages to accumulate in the same bat Default: `0`. messageKeyExpression:: A SpEL expression evaluated against the outgoing message used to populate the key of the produced Kafka message -- for example, `headers['myKey']`. -The payload cannot be used because, by the time this expression is evaluated, the payload is already in the form of a `byte[]`. +With versions before 3.0, the payload could not be used unless native encoding was being used because, by the time this expression was evaluated, the payload was already in the form of a `byte[]`. +Now, the expression is evaluated before the payload is converted. + Default: `none`. headerPatterns:: @@ -341,7 +348,21 @@ The replication factor to use when provisioning topics. Overrides the binder-wid Ignored if `replicas-assignments` is present. + Default: none (the binder-wide default of 1 is used). +useTopicHeader:: +Set to `true` to override the default binding destination (topic name) with the value of the `KafkaHeaders.TOPIC` message header in the outbound message. +If the header is not present, the default binding destination is used. +Default: `false`. ++ +recordMetadataChannel:: +The bean name of a `MessageChannel` to which successful send results should be sent; the bean must exist in the application context. +The message sent to the channel is the sent message (after conversion, if any) with an additional header `KafkaHeaders.RECORD_METADATA`. +The header contains a `RecordMetadata` object provided by the Kafka client; it includes the partition and offset where the record was written in the topic. +`ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)` + +Failed sends go the producer error channel (if configured); see <>. +Default: null ++ NOTE: The Kafka binder uses the `partitionCount` setting of the producer as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used). Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value is used. @@ -516,6 +537,50 @@ public class Application { } ---- +[[kafka-transactional-binder]] +=== Transactional Binder + +Enable transactions by setting `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` to a non-empty value, e.g. `tx-`. +When used in a processor application, the consumer starts the transaction; any records sent on the consumer thread participate in the same transaction. +When the listener exits normally, the listener container will send the offset to the transaction and commit it. +A common producer factory is used for all producer bindings configured using `spring.cloud.stream.kafka.binder.transaction.producer.*` properties; individual binding Kafka producer properties are ignored. + +If you wish to use transactions in a source application, or from some arbitrary thread for producer-only transaction (e.g. `@Scheduled` method), you must get a reference to the transactional producer factory and define a `KafkaTransactionManager` bean using it. + +==== +[source, java] +---- +@Bean +public PlatformTransactionManager transactionManager(BinderFactory binders) { + ProducerFactory pf = ((KafkaMessageChannelBinder) binders.getBinder(null, + MessageChannel.class)).getTransactionalProducerFactory(); + return new KafkaTransactionManager<>(pf); +} +---- +==== + +Notice that we get a reference to the binder using the `BinderFactory`; use `null` in the first argument when there is only one binder configured. +If more than one binder is configured, use the binder name to get the reference. +Once we have a reference to the binder, we can obtain a reference to the `ProducerFactory` and create a transaction manager. + +Then you would use normal Spring transaction support, e.g. `TransactionTemplate` or `@Transactional`, for example: + +==== +[source, java] +---- +public static class Sender { + + @Transactional + public void doInTransaction(MessageChannel output, List stuffToSend) { + stuffToSend.forEach(stuff -> output.send(new GenericMessage<>(stuff))); + } + +} +---- +==== + +If you wish to synchronize producer-only transactions with those from some other transaction manager, use a `ChainedTransactionManager`. + [[kafka-error-channels]] === Error Channels @@ -731,4 +796,4 @@ added after the original pull request but before a merge. if you are fixing an existing issue please add `Fixes gh-XXXX` at the end of the commit message (where XXXX is the issue number). -// ====================================================================================== +// ====================================================================================== \ No newline at end of file diff --git a/docs/pom.xml b/docs/pom.xml index 5ace9406d..1354f8d84 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -15,311 +15,50 @@ spring-cloud-stream-binder-kafka ${basedir}/.. - 0.1.1.RELEASE - 0.1.0.RELEASE - - 1.5.0-alpha.16 + 3.4 + + + + maven-deploy-plugin + 2.8.2 + + true + + + + docs + + pl.project13.maven + git-commit-id-plugin + org.apache.maven.plugins maven-dependency-plugin - ${maven-dependency-plugin.version} - false - - - unpack-docs - generate-resources - - unpack - - - - - org.springframework.cloud - - spring-cloud-build-docs - - ${spring-cloud-build.version} - - sources - jar - false - ${docs.resources.dir} - - - - - - - unpack-docs-resources - generate-resources - - unpack - - - - - io.spring.docresources - spring-doc-resources - ${spring-doc-resources.version} - zip - true - ${project.build.directory}/refdocs/ - - - - - org.apache.maven.plugins maven-resources-plugin - - - copy-asciidoc-resources - generate-resources - - copy-resources - - - ${project.build.directory}/refdocs/ - - - src/main/asciidoc - false - - ghpages.sh - - - - - - - org.asciidoctor asciidoctor-maven-plugin ${asciidoctor-maven-plugin.version} - false - - - io.spring.asciidoctor - spring-asciidoctor-extensions - ${spring-asciidoctor-extensions.version} - - - org.asciidoctor - asciidoctorj-pdf - ${asciidoctorj-pdf.version} - - ${project.build.directory}/refdocs/ - ${docs.main}.adoc - ${project.version} - - - ${project.version}/ - https://cloud.spring.io/spring-cloud-static/ - + ${project.version} + - - - generate-html-documentation - prepare-package - - process-asciidoc - - - html5 - highlight.js - book - - // these attributes are required to use the doc resources - shared - css/ - spring.css - true - font - js/highlight - atom-one-dark-reasonable - true - - left - 4 - ${project.version} - true - - ${docs.main}.html - - - - generate-docbook - none - - process-asciidoc - - - - generate-index - none - - process-asciidoc - - - - - org.apache.maven.plugins maven-antrun-plugin - ${maven-antrun-plugin.version} - - - ant-contrib - ant-contrib - 1.0b3 - - - ant - ant - - - - - org.apache.ant - ant-nodeps - 1.8.1 - - - org.tigris.antelope - antelopetasks - 3.2.10 - - - org.jruby - jruby-complete - 1.7.17 - - - org.asciidoctor - asciidoctorj - 1.5.8 - - - - - readme - process-resources - - run - - - - - - - - - - - - - assert-no-unresolved-links - prepare-package - - run - - - - - - - - - - - - - - - - setup-maven-properties - validate - - run - - - true - - - - - - - - - - - - - - - - - - copy-css - none - - run - - - - generate-documentation-index - none - - run - - - - copy-generated-html - none - - run - - - - - - - - org.codehaus.mojo - build-helper-maven-plugin - false diff --git a/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc b/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc index cf1331958..5ddd850ab 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream-binder-kafka.adoc @@ -34,8 +34,6 @@ Sabby Anandan, Marius Bogoevici, Eric Bottard, Mark Fisher, Ilayaperumal Gopinat *{spring-cloud-stream-version}* -[#index-link] -{docs-url}spring-cloud-stream/{docs-version}home.html = Reference Guide include::overview.adoc[]