diff --git a/README.adoc b/README.adoc index 155660523..2c8936749 100644 --- a/README.adoc +++ b/README.adoc @@ -210,6 +210,7 @@ If not set (the default), it effectively has the same value as `enableDlq`, auto Default: not set. resetOffsets:: Whether to reset offsets on the consumer to the value provided by startOffset. +Must be false if a `KafkaRebalanceListener` is provided; see <>. + Default: `false`. startOffset:: @@ -531,6 +532,57 @@ public void in(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) byte[] key, ---- ==== +[[rebalance-listener]] +== Using a KafkaRebalanceListener + +Applications may wish to seek topics/partitions to arbitrary offsets when the partitions are initially assigned, or perform other operations on the consumer. +Starting with version 2.1, if you provide a single `KafkaRebalanceListener` bean in the application context, it will be wired into all Kafka consumer bindings. + +==== +[source, java] +---- +public interface KafkaBindingRebalanceListener { + + /** + * Invoked by the container before any pending offsets are committed. + * @param bindingName the name of the binding. + * @param consumer the consumer. + * @param partitions the partitions. + */ + default void onPartitionsRevokedBeforeCommit(String bindingName, Consumer consumer, + Collection partitions) { + + } + + /** + * Invoked by the container after any pending offsets are committed. + * @param bindingName the name of the binding. + * @param consumer the consumer. + * @param partitions the partitions. + */ + default void onPartitionsRevokedAfterCommit(String bindingName, Consumer consumer, Collection partitions) { + + } + + /** + * Invoked when partitions are initially assigned or after a rebalance. + * Applications might only want to perform seek operations on an initial assignment. + * @param bindingName the name of the binding. + * @param consumer the consumer. + * @param partitions the partitions. + * @param initial true if this is the initial assignment. + */ + default void onPartitionsAssigned(String bindingName, Consumer consumer, Collection partitions, + boolean initial) { + + } + +} +---- +==== + +You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener. + = Appendices [appendix] [[building]] diff --git a/docs/pom.xml b/docs/pom.xml index 42e3c7272..3b659152d 100644 --- a/docs/pom.xml +++ b/docs/pom.xml @@ -7,7 +7,7 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-parent - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2 pom spring-cloud-stream-binder-kafka-docs diff --git a/pom.xml b/pom.xml index 320947bd8..907857f68 100644 --- a/pom.xml +++ b/pom.xml @@ -2,12 +2,12 @@ 4.0.0 spring-cloud-stream-binder-kafka-parent - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2 pom org.springframework.cloud spring-cloud-build - 2.1.0.RC1 + 2.1.0.RC2 @@ -15,7 +15,7 @@ 2.2.0.RELEASE 3.1.0.RELEASE 2.0.0 - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2 spring-cloud-stream-binder-kafka diff --git a/spring-cloud-starter-stream-kafka/pom.xml b/spring-cloud-starter-stream-kafka/pom.xml index 201bf41dd..ce65dbc57 100644 --- a/spring-cloud-starter-stream-kafka/pom.xml +++ b/spring-cloud-starter-stream-kafka/pom.xml @@ -4,7 +4,7 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-parent - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2 spring-cloud-starter-stream-kafka Spring Cloud Starter Stream Kafka diff --git a/spring-cloud-stream-binder-kafka-core/pom.xml b/spring-cloud-stream-binder-kafka-core/pom.xml index 691e24a6c..2e5a7fcf7 100644 --- a/spring-cloud-stream-binder-kafka-core/pom.xml +++ b/spring-cloud-stream-binder-kafka-core/pom.xml @@ -5,7 +5,7 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-parent - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2 spring-cloud-stream-binder-kafka-core Spring Cloud Stream Kafka Binder Core diff --git a/spring-cloud-stream-binder-kafka-streams/pom.xml b/spring-cloud-stream-binder-kafka-streams/pom.xml index c2be860b0..356580c0b 100644 --- a/spring-cloud-stream-binder-kafka-streams/pom.xml +++ b/spring-cloud-stream-binder-kafka-streams/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-parent - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2 diff --git a/spring-cloud-stream-binder-kafka/pom.xml b/spring-cloud-stream-binder-kafka/pom.xml index 6fe82ddf0..30a4e10e4 100644 --- a/spring-cloud-stream-binder-kafka/pom.xml +++ b/spring-cloud-stream-binder-kafka/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-kafka-parent - 2.1.0.BUILD-SNAPSHOT + 2.1.0.RC2