This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.2.x
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/1.2.x by this push:
new d108ba41 Clear lastRevoked after use to avoid wiping buffers under
cooperative rebalancing (#632)
d108ba41 is described below
commit d108ba41a6340abbc69caf242bb38cd128b1695b
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 17 10:51:00 2026 +0100
Clear lastRevoked after use to avoid wiping buffers under cooperative
rebalancing (#632)
* Clear lastRevoked after use to avoid wiping buffers under cooperative
rebalancing (#617)
* Clear lastRevoked after use to avoid wiping buffers under cooperative
rebalancing
* With the cooperative rebalance protocol onPartitionsRevoked is only
invoked on
members that actually revoke partitions, while onPartitionsAssigned is
invoked
on every member in every rebalance. A stale lastRevoked set from an
earlier
rebalance was re-applied on such assignment callbacks and filtered
buffered
records of partitions that remained assigned, losing records without them
being processed or committed.
* Reset lastRevoked after the flush in SingleSourceLogic and SubSourceLogic.
* Run the RebalanceSpec buffer tests against both the eager and cooperative
protocols using a cooperative variant of the test assignor.
* Add CooperativeRebalanceSpec reproducing the record loss for plain and
partitioned sources; both tests fail without the fix.
Fixes #616
Co-Authored-By: Claude Fable 5 <[email protected]>
* Address review: standard ASF header for new file, drop final vals and
test logging, apply scalafmt
Co-Authored-By: Claude Fable 5 <[email protected]>
---------
Co-authored-by: Claude Fable 5 <[email protected]>
* merge issue
* upgrade CI actions
* Remove pull_request trigger from link-validator workflow
* Add case to handle unexpected messages in rebalance test
---------
Co-authored-by: Arvind Raghavan <[email protected]>
Co-authored-by: Claude Fable 5 <[email protected]>
---
.github/workflows/check-build-test.yml | 46 +--
.github/workflows/format.yml | 4 +-
.github/workflows/headers.yml | 8 +-
.github/workflows/link-validator.yml | 11 +-
.github/workflows/nightly-pekko.yml | 20 +-
.github/workflows/publish-1.0-docs.yml | 8 +-
.github/workflows/publish-1.1-docs.yml | 8 +-
.github/workflows/publish-nightly.yml | 8 +-
.../pekko/kafka/internal/SingleSourceLogic.scala | 4 +-
.../pekko/kafka/internal/SubSourceLogic.scala | 4 +-
.../kafka/scaladsl/CooperativeRebalanceSpec.scala | 175 ++++++++
.../pekko/kafka/scaladsl/RebalanceSpec.scala | 441 ++++++++++++---------
12 files changed, 497 insertions(+), 240 deletions(-)
diff --git a/.github/workflows/check-build-test.yml
b/.github/workflows/check-build-test.yml
index 4185ae2d..0f17f4b0 100644
--- a/.github/workflows/check-build-test.yml
+++ b/.github/workflows/check-build-test.yml
@@ -24,7 +24,7 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
# we don't know what commit the last tag was it's safer to get
entire repo so previousStableVersion resolves
fetch-depth: 0
@@ -38,16 +38,16 @@ jobs:
- name: Setup Java 11
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 11
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Code style check and binary-compatibility check
# Run locally with: sbt 'javafmtCheckAll; mimaReportBinaryIssues'
@@ -58,7 +58,7 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
@@ -70,13 +70,13 @@ jobs:
git checkout scratch
- name: Setup Java 11
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 11
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Compile all code with fatal warnings for Java 11 and Scala
2.12/2.13/3
# Run locally with: env CI=true sbt 'clean ; Test/compile'
@@ -87,7 +87,7 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
@@ -99,16 +99,16 @@ jobs:
git checkout scratch
- name: Setup Java 11
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 11
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Create all API docs for artifacts/website and all reference docs
run: sbt "doc ; unidoc ; docs/paradoxBrowse"
@@ -129,7 +129,7 @@ jobs:
- { java-version: 11, scala-version: 3.3, sbt-opts:
'-J-XX:+UnlockExperimentalVMOptions -J-XX:+UseJVMCICompiler' }
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
@@ -141,16 +141,16 @@ jobs:
git checkout scratch
- name: Setup Java ${{ matrix.java-version }}
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: ${{ matrix.java-version }}
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Run tests with Scala ${{ matrix.scala-version }} and Java ${{
matrix.java-version }}
run: sbt "++${{ matrix.scala-version }} test" ${{ matrix.sbt-opts }}
@@ -165,7 +165,7 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
@@ -177,16 +177,16 @@ jobs:
git checkout scratch
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Run multi-broker and long running integration tests
run: sbt "int-tests/test"
@@ -200,7 +200,7 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
@@ -212,16 +212,16 @@ jobs:
git checkout scratch
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: "Compile benchmark tests"
run: sbt "benchmarks/Test/compile"
diff --git a/.github/workflows/format.yml b/.github/workflows/format.yml
index f31104de..64f08b27 100644
--- a/.github/workflows/format.yml
+++ b/.github/workflows/format.yml
@@ -12,12 +12,12 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout current branch (full)
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
persist-credentials: false
- name: Check project is formatted
- uses: jrouly/scalafmt-native-action@v4
+ uses: jrouly/scalafmt-native-action@v5
with:
arguments: '--list --mode diff-ref=origin/main'
diff --git a/.github/workflows/headers.yml b/.github/workflows/headers.yml
index 830d0f80..3c4ce202 100644
--- a/.github/workflows/headers.yml
+++ b/.github/workflows/headers.yml
@@ -11,19 +11,19 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
- name: Setup Java 11
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 11
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Check headers
run: |-
diff --git a/.github/workflows/link-validator.yml
b/.github/workflows/link-validator.yml
index 96479f71..aeba0f8f 100644
--- a/.github/workflows/link-validator.yml
+++ b/.github/workflows/link-validator.yml
@@ -3,7 +3,6 @@ name: Link Validator
permissions: {}
on:
- pull_request:
workflow_dispatch:
schedule:
- cron: '0 6 * * 1'
@@ -13,25 +12,25 @@ jobs:
runs-on: ubuntu-22.04
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Setup Coursier
- uses: coursier/setup-action@039f736548afa5411c1382f40a5bd9c2d30e0383 #
v1.3.9
+ uses: coursier/setup-action@fd1707a76b027efdfb66ca79318b4d29b72e5a02 #
v3.0.0
- name: Build Documentation
run: |-
diff --git a/.github/workflows/nightly-pekko.yml
b/.github/workflows/nightly-pekko.yml
index e57d94c2..48947201 100644
--- a/.github/workflows/nightly-pekko.yml
+++ b/.github/workflows/nightly-pekko.yml
@@ -27,25 +27,25 @@ jobs:
PEKKO_VERSION: ['main', '1.0.x']
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
- name: Set up JDK ${{ matrix.JDK }}
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: ${{ matrix.JDK }}
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Cache Build Target
- uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
+ uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
with:
path: project/**/target
key: build-target-${{ hashFiles('**/*.sbt',
'project/build.properties', 'project/**/*.scala') }}
@@ -68,25 +68,25 @@ jobs:
PEKKO_VERSION: ['main', '1.0.x']
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Cache Build Target
- uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
+ uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
with:
path: project/**/target
key: build-target-${{ hashFiles('**/*.sbt',
'project/build.properties', 'project/**/*.scala') }}
diff --git a/.github/workflows/publish-1.0-docs.yml
b/.github/workflows/publish-1.0-docs.yml
index 95d5bda9..8199c7e3 100644
--- a/.github/workflows/publish-1.0-docs.yml
+++ b/.github/workflows/publish-1.0-docs.yml
@@ -10,23 +10,23 @@ jobs:
if: github.repository == 'apache/pekko-connectors-kafka'
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
ref: 1.0.x
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Build Documentation
run: |-
diff --git a/.github/workflows/publish-1.1-docs.yml
b/.github/workflows/publish-1.1-docs.yml
index 3c542fe9..bed9e19a 100644
--- a/.github/workflows/publish-1.1-docs.yml
+++ b/.github/workflows/publish-1.1-docs.yml
@@ -10,22 +10,22 @@ jobs:
if: github.repository == 'apache/pekko-connectors-kafka'
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Build Documentation
run: |-
diff --git a/.github/workflows/publish-nightly.yml
b/.github/workflows/publish-nightly.yml
index de7704d5..ec014d14 100644
--- a/.github/workflows/publish-nightly.yml
+++ b/.github/workflows/publish-nightly.yml
@@ -32,19 +32,19 @@ jobs:
if: github.repository == 'apache/pekko-connectors-kafka'
steps:
- name: Checkout
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 #
v4.2.2
+ uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 #
v6.0.3
with:
fetch-depth: 0
fetch-tags: true
- name: Setup Java 8
- uses: actions/setup-java@c5195efecf7bdfc987ee8bae7a71cb8b11521c00 #
v4.7.1
+ uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 #
v5.2.0
with:
distribution: temurin
java-version: 8
- name: Install sbt
- uses: sbt/setup-sbt@6c68d2fe8dfbc0a0534d70101baa2e0420e1a506 # v1.1.9
+ uses: sbt/setup-sbt@bfea3c5f48abd221b04a6df4798aa5eb8b6a2baf # v1.5.6
- name: Publish to Apache Maven repo
env:
@@ -53,7 +53,7 @@ jobs:
run: sbt +publish
- name: Cache Coursier cache
- uses: coursier/cache-action@4e2615869d13561d626ed48655e1a39e5b192b3c #
v6.4.7
+ uses: coursier/cache-action@95e5b1029b6b86e7bac033ee44a0697d8a527d2d #
8.1.1
- name: Build Documentation
run: |-
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
index c7726ca7..9a36b884 100644
---
a/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
+++
b/core/src/main/scala/org/apache/pekko/kafka/internal/SingleSourceLogic.scala
@@ -96,8 +96,10 @@ import scala.concurrent.{ Future, Promise }
override def onRevoke(revokedTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit =
lastRevoked = revokedTps
- override def onAssign(assignedTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit =
+ override def onAssign(assignedTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit = {
filterRevokedPartitionsCB.invoke(lastRevoked -- assignedTps)
+ lastRevoked = Set.empty
+ }
override def onLost(lostTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit =
filterRevokedPartitionsCB.invoke(lostTps)
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
index dff06e84..8000a0d1 100644
--- a/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
+++ b/core/src/main/scala/org/apache/pekko/kafka/internal/SubSourceLogic.scala
@@ -301,11 +301,13 @@ private class SubSourceLogic[K, V, Msg](
override def onRevoke(revokedTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit =
lastRevoked = revokedTps
- override def onAssign(assignedTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit =
+ override def onAssign(assignedTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit = {
for {
tp <- lastRevoked -- assignedTps
control <- subSources.get(tp)
} control.filterRevokedPartitionsCB.invoke(Set(tp))
+ lastRevoked = Set.empty
+ }
override def onLost(lostTps: Set[TopicPartition], consumer:
RestrictedConsumer): Unit =
for {
diff --git
a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CooperativeRebalanceSpec.scala
b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CooperativeRebalanceSpec.scala
new file mode 100644
index 00000000..d2a8995f
--- /dev/null
+++
b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/CooperativeRebalanceSpec.scala
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.kafka.scaladsl
+
+import org.apache.pekko
+import pekko.Done
+import pekko.kafka._
+import pekko.kafka.scaladsl.Consumer.Control
+import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.stream.scaladsl.{ Keep, Source }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestProbe
+import org.apache.kafka.clients.consumer.{ ConsumerConfig, ConsumerRecord }
+import org.apache.kafka.common.TopicPartition
+import org.scalatest.Inside
+
+import scala.concurrent.duration._
+import scala.util.Random
+
+/**
+ * With the cooperative rebalance protocol `onPartitionsRevoked` is only
invoked on members that
+ * actually revoke partitions, while `onPartitionsAssigned` is invoked on
every member in every
+ * rebalance, possibly with an empty set. These tests reproduce rebalances
where the revoke
+ * callback is skipped and assert that buffered records of partitions that
remain assigned are
+ * still delivered (at-least-once).
+ *
+ * Reproduces the scenario of
https://github.com/apache/pekko-connectors-kafka/issues/616:
+ * 1. a partition is revoked from consumer 1 (its `lastRevoked` state becomes
non-empty),
+ * 2. the partition is later re-assigned to consumer 1,
+ * 3. an unrelated member joins, completing a rebalance in which consumer 1
revokes nothing
+ * (`onPartitionsRevoked` is not invoked) and gains nothing
(`onPartitionsAssigned` with an
+ * empty set) while records for the re-assigned partition sit in the
source stage buffer.
+ *
+ * The test runs against `plainSource` and `plainPartitionedSource` (flattened
with
+ * `flatMapMerge` so both shapes expose a single stream of records) as the
underlying stages
+ * implement the revoked-buffer bookkeeping independently. The merge stage of
the flattened
+ * variant buffers a few records of its own that survive a revoke as
duplicates, so the tests
+ * assert complete delivery rather than an exact sequence.
+ */
+class CooperativeRebalanceSpec extends SpecBase with TestcontainersKafkaLike
with Inside {
+
+ implicit override val patienceConfig: PatienceConfig =
PatienceConfig(30.seconds, 500.millis)
+
+ val partition1 = 1
+ val consumerClientId1 = "consumer-1"
+ val consumerClientId2 = "consumer-2"
+ val consumerClientId3 = "consumer-3"
+
+ private def cooperativeSettings(group: String) =
+ consumerDefaults
+ .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500")
+ .withProperty(
+ ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
+ classOf[CooperativePekkoConnectorsAssignor].getName)
+ .withGroupId(group)
+
+ private def awaitAssigned(rebalanceActor: TestProbe,
+ subscription: AutoSubscription,
+ tps: Set[TopicPartition]): Unit =
+ rebalanceActor.fishForMessage(10.seconds) {
+ case TopicPartitionsAssigned(`subscription`, assigned) if assigned ==
tps => true
+ case TopicPartitionsAssigned(`subscription`, assigned) if
assigned.isEmpty => false
+ case _
=> false
+ }
+
+ sealed trait SourceCase {
+ def label: String
+ def source(settings: ConsumerSettings[String, String],
+ subscription: AutoSubscription): Source[ConsumerRecord[String,
String], Control]
+ }
+
+ case object PlainCase extends SourceCase {
+ override val label = "plain source"
+ override def source(settings: ConsumerSettings[String, String],
+ subscription: AutoSubscription): Source[ConsumerRecord[String,
String], Control] =
+ Consumer.plainSource(settings, subscription)
+ }
+
+ case object PartitionedCase extends SourceCase {
+ override val label = "partitioned source"
+ override def source(settings: ConsumerSettings[String, String],
+ subscription: AutoSubscription): Source[ConsumerRecord[String,
String], Control] =
+ Consumer
+ .plainPartitionedSource(settings, subscription)
+ .flatMapMerge(breadth = 8, { case (_, records) => records })
+ }
+
+ "Buffered records of partitions that stay assigned" must {
+
+ List(PlainCase, PartitionedCase).foreach { mode =>
+ s"be delivered after a rebalance without revocation (${mode.label})" in
assertAllStagesStopped {
+ val count = 100L
+ val topicSuffix = Random.nextInt()
+ val topic1 = createTopic(topicSuffix, partitions = 2)
+ val group1 = createGroupId(1)
+ val tp0 = new TopicPartition(topic1, partition0)
+ val tp1 = new TopicPartition(topic1, partition1)
+ val consumerSettings = cooperativeSettings(group1)
+
+ def joinConsumer(clientId: String) = {
+ val rebalanceActor = TestProbe()
+ val subscription =
Subscriptions.topics(topic1).withRebalanceListener(rebalanceActor.ref)
+ val (control, probe) = Consumer
+ .plainSource(consumerSettings.withClientId(clientId), subscription)
+ .toMat(TestSink())(Keep.both)
+ .run()
+ (control, probe, rebalanceActor, subscription)
+ }
+
+ awaitProduce(produce(topic1, 0 to count.toInt, partition1))
+
+
PekkoConnectorsAssignor.clientIdToPartitionMap.set(Map(consumerClientId1 ->
Set(tp0, tp1)))
+
+ val probe1rebalanceActor = TestProbe()
+ val probe1subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe1rebalanceActor.ref)
+ val (control1, probe1) = mode
+ .source(consumerSettings.withClientId(consumerClientId1),
probe1subscription)
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+
probe1rebalanceActor.expectMsg(TopicPartitionsAssigned(probe1subscription,
Set(tp0, tp1)))
+ probe1.requestNext()
+
+ PekkoConnectorsAssignor.clientIdToPartitionMap.set(
+ Map(consumerClientId1 -> Set(tp0), consumerClientId2 -> Set(tp1)))
+ val (control2, probe2, probe2rebalanceActor, probe2subscription) =
joinConsumer(consumerClientId2)
+
+
probe1rebalanceActor.expectMsg(TopicPartitionsRevoked(probe1subscription,
Set(tp1)))
+ awaitAssigned(probe2rebalanceActor, probe2subscription, Set(tp1))
+
+
PekkoConnectorsAssignor.clientIdToPartitionMap.set(Map(consumerClientId1 ->
Set(tp0, tp1)))
+ probe2.cancel()
+ control2.isShutdown.futureValue shouldBe Done
+ awaitAssigned(probe1rebalanceActor, probe1subscription, Set(tp1))
+
+ probe1.requestNext()
+
+ PekkoConnectorsAssignor.clientIdToPartitionMap.set(
+ Map(consumerClientId1 -> Set(tp0, tp1), consumerClientId3 ->
Set.empty[TopicPartition]))
+ val (control3, probe3, probe3rebalanceActor, probe3subscription) =
joinConsumer(consumerClientId3)
+
+
probe1rebalanceActor.expectMsg(TopicPartitionsAssigned(probe1subscription,
Set.empty))
+
probe3rebalanceActor.expectMsg(TopicPartitionsAssigned(probe3subscription,
Set.empty))
+
+ // give the asynchronous buffer filter of the rebalance a chance to
apply before demanding
+ probe1.expectNoMessage(500.millis)
+
+ probe1.request(count * 3)
+ val values = probe1.receiveWithin(5.seconds).map(_.value)
+ values should contain allElementsOf (1 to count.toInt).map(_.toString)
+
+ probe1.cancel()
+ probe3.cancel()
+ control1.isShutdown.futureValue shouldBe Done
+ control3.isShutdown.futureValue shouldBe Done
+ }
+ }
+ }
+}
diff --git
a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
index a673428a..a8c8e2b9 100644
--- a/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
+++ b/tests/src/test/scala/org/apache/pekko/kafka/scaladsl/RebalanceSpec.scala
@@ -20,13 +20,13 @@ import java.util.concurrent.atomic.AtomicReference
import org.apache.pekko
import pekko.kafka._
import pekko.kafka.testkit.scaladsl.TestcontainersKafkaLike
+import pekko.util.ccompat.JavaConverters._
import pekko.stream.scaladsl.{ Keep, Source }
import pekko.stream.testkit.TestSubscriber
import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import pekko.stream.testkit.scaladsl.TestSink
import pekko.testkit.TestProbe
import pekko.{ Done, NotUsed }
-import pekko.util.ccompat.JavaConverters._
import org.apache.kafka.clients.consumer.{ ConsumerConfig,
ConsumerPartitionAssignor, ConsumerRecord }
import org.apache.kafka.clients.consumer.internals.AbstractPartitionAssignor
import org.apache.kafka.common.TopicPartition
@@ -45,197 +45,241 @@ class RebalanceSpec extends SpecBase with
TestcontainersKafkaLike with Inside {
final val consumerClientId1 = "consumer-1"
final val consumerClientId2 = "consumer-2"
+ sealed trait ProtocolCase {
+ def label: String
+ def assignor: Class[?]
+
+ /** Expected rebalance listener events after the second consumer joins and
takes tp1. */
+ def expectSecondConsumerJoin(
+ probe1rebalanceActor: TestProbe,
+ probe1subscription: AutoSubscription,
+ probe2rebalanceActor: TestProbe,
+ probe2subscription: AutoSubscription,
+ tp0: TopicPartition,
+ tp1: TopicPartition): Unit
+ }
+
+ case object EagerCase extends ProtocolCase {
+ override val label = "eager"
+ override val assignor: Class[?] = classOf[PekkoConnectorsAssignor]
+
+ override def expectSecondConsumerJoin(
+ probe1rebalanceActor: TestProbe,
+ probe1subscription: AutoSubscription,
+ probe2rebalanceActor: TestProbe,
+ probe2subscription: AutoSubscription,
+ tp0: TopicPartition,
+ tp1: TopicPartition): Unit = {
+
probe1rebalanceActor.expectMsg(TopicPartitionsRevoked(probe1subscription,
Set(tp0, tp1)))
+
probe1rebalanceActor.expectMsg(TopicPartitionsAssigned(probe1subscription,
Set(tp0)))
+
probe2rebalanceActor.expectMsg(TopicPartitionsAssigned(probe2subscription,
Set(tp1)))
+ }
+ }
+
+ case object CooperativeCase extends ProtocolCase {
+ override val label = "cooperative"
+ override val assignor: Class[?] =
classOf[CooperativePekkoConnectorsAssignor]
+
+ override def expectSecondConsumerJoin(
+ probe1rebalanceActor: TestProbe,
+ probe1subscription: AutoSubscription,
+ probe2rebalanceActor: TestProbe,
+ probe2subscription: AutoSubscription,
+ tp0: TopicPartition,
+ tp1: TopicPartition): Unit = {
+ // tp1 is revoked from consumer 1 and assigned to consumer 2 in a
follow-up rebalance.
+ // Intermediate empty assignments are not asserted: members that fail
the intermediate
+ // generation's sync with REBALANCE_IN_PROGRESS skip its assignment
callback entirely.
+
probe1rebalanceActor.expectMsg(TopicPartitionsRevoked(probe1subscription,
Set(tp1)))
+ probe2rebalanceActor.fishForMessage(10.seconds) {
+ case TopicPartitionsAssigned(`probe2subscription`, assigned) if
assigned == Set(tp1) => true
+ case TopicPartitionsAssigned(`probe2subscription`, assigned) if
assigned.isEmpty => false
+ }
+ }
+ }
+
"Fetched records" must {
// The `max.poll.records` controls how many records Kafka fetches
internally during a poll.
// issue explained in https://github.com/akka/alpakka-kafka/issues/872
// this test added with https://github.com/akka/alpakka-kafka/pull/865
- "be removed from the source stage buffer when a partition is revoked" in
assertAllStagesStopped {
- val count = 20L
- // de-coupling consecutive test runs with crossScalaVersions on build
- val topicSuffix = Random.nextInt()
- val topic1 = createTopic(topicSuffix, partitions = 2)
- val group1 = createGroupId(1)
- val tp0 = new TopicPartition(topic1, partition0)
- val tp1 = new TopicPartition(topic1, partition1)
- val consumerSettings = consumerDefaults
- .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500") // 500 is
the default value
- .withProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
classOf[PekkoConnectorsAssignor].getName)
- .withGroupId(group1)
-
- awaitProduce(produce(topic1, 0 to count.toInt, partition1))
-
- PekkoConnectorsAssignor.clientIdToPartitionMap.set(
- Map(
- consumerClientId1 -> Set(tp0, tp1)))
-
- log.debug("Subscribe to the topic (without demand)")
- val probe1rebalanceActor = TestProbe()
- val probe1subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe1rebalanceActor.ref)
- val (control1, probe1) = Consumer
- .plainSource(consumerSettings.withClientId(consumerClientId1),
probe1subscription)
- .toMat(TestSink.probe)(Keep.both)
- .run()
-
- log.debug("Await initial partition assignment")
- probe1rebalanceActor.expectMsg(
- TopicPartitionsAssigned(probe1subscription,
- Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
-
- log.debug("read one message from probe1 with partition 1")
- probe1.requestNext()
-
- PekkoConnectorsAssignor.clientIdToPartitionMap.set(
- Map(
- consumerClientId1 -> Set(tp0),
- consumerClientId2 -> Set(tp1)))
-
- log.debug("Subscribe to the topic (without demand)")
- val probe2rebalanceActor = TestProbe()
- val probe2subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe2rebalanceActor.ref)
- val (control2, probe2) = Consumer
- .plainSource(consumerSettings.withClientId(consumerClientId2),
probe2subscription)
- .toMat(TestSink.probe)(Keep.both)
- .run()
-
- log.debug("Await a revoke to consumer 1")
- probe1rebalanceActor.expectMsg(
- TopicPartitionsRevoked(probe1subscription,
- Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
-
- log.debug("the rebalance finishes")
- probe1rebalanceActor.expectMsg(
- TopicPartitionsAssigned(probe1subscription, Set(new
TopicPartition(topic1, partition0))))
- probe2rebalanceActor.expectMsg(
- TopicPartitionsAssigned(probe2subscription, Set(new
TopicPartition(topic1, partition1))))
-
- log.debug("resume demand on both consumers")
- probe1.request(count)
- probe2.request(count)
-
- val probe2messages = probe2.expectNextN(count)
-
- log.debug("no further messages enqueued on probe1 as partition 1 is
balanced away")
- probe1.expectNoMessage(500.millis)
-
- probe2messages should have size count
-
- probe1.cancel()
- probe2.cancel()
-
- control1.isShutdown.futureValue shouldBe Done
- control2.isShutdown.futureValue shouldBe Done
+ List(EagerCase, CooperativeCase).foreach { mode =>
+ s"be removed from the source stage buffer when a partition is revoked
(${mode.label})" in assertAllStagesStopped {
+ val count = 20L
+ // de-coupling consecutive test runs with crossScalaVersions on build
+ val topicSuffix = Random.nextInt()
+ val topic1 = createTopic(topicSuffix, partitions = 2)
+ val group1 = createGroupId(1)
+ val tp0 = new TopicPartition(topic1, partition0)
+ val tp1 = new TopicPartition(topic1, partition1)
+ val consumerSettings = consumerDefaults
+ .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500") // 500
is the default value
+ .withProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
mode.assignor.getName)
+ .withGroupId(group1)
+
+ awaitProduce(produce(topic1, 0 to count.toInt, partition1))
+
+ PekkoConnectorsAssignor.clientIdToPartitionMap.set(
+ Map(
+ consumerClientId1 -> Set(tp0, tp1)))
+
+ log.debug("Subscribe to the topic (without demand)")
+ val probe1rebalanceActor = TestProbe()
+ val probe1subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe1rebalanceActor.ref)
+ val (control1, probe1) = Consumer
+ .plainSource(consumerSettings.withClientId(consumerClientId1),
probe1subscription)
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ log.debug("Await initial partition assignment")
+ probe1rebalanceActor.expectMsg(
+ TopicPartitionsAssigned(probe1subscription,
+ Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
+
+ log.debug("read one message from probe1 with partition 1")
+ probe1.requestNext()
+
+ PekkoConnectorsAssignor.clientIdToPartitionMap.set(
+ Map(
+ consumerClientId1 -> Set(tp0),
+ consumerClientId2 -> Set(tp1)))
+
+ log.debug("Subscribe to the topic (without demand)")
+ val probe2rebalanceActor = TestProbe()
+ val probe2subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe2rebalanceActor.ref)
+ val (control2, probe2) = Consumer
+ .plainSource(consumerSettings.withClientId(consumerClientId2),
probe2subscription)
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ log.debug("Await the rebalance to complete")
+ mode.expectSecondConsumerJoin(probe1rebalanceActor,
probe1subscription, probe2rebalanceActor,
+ probe2subscription, tp0, tp1)
+
+ log.debug("resume demand on both consumers")
+ probe1.request(count)
+ probe2.request(count)
+
+ val probe2messages = probe2.expectNextN(count)
+
+ log.debug("no further messages enqueued on probe1 as partition 1 is
balanced away")
+ probe1.expectNoMessage(500.millis)
+
+ probe2messages should have size count
+
+ probe1.cancel()
+ probe2.cancel()
+
+ control1.isShutdown.futureValue shouldBe Done
+ control2.isShutdown.futureValue shouldBe Done
+ }
}
- "be removed from the partitioned source stage buffer when a partition is
revoked" in assertAllStagesStopped {
- def subSourcesWithProbes(
- partitions: Int,
- probe: TestSubscriber.Probe[(TopicPartition,
Source[ConsumerRecord[String, String], NotUsed])])
- : Seq[(TopicPartition, TestSubscriber.Probe[ConsumerRecord[String,
String]])] =
- probe
- .expectNextN(partitions.toLong)
- .map {
- case (tp, subSource) =>
- (tp, subSource.toMat(TestSink.probe)(Keep.right).run())
+ List(EagerCase, CooperativeCase).foreach { mode =>
+ s"be removed from the partitioned source stage buffer when a partition
is revoked (${mode.label})" in
+ assertAllStagesStopped {
+ def subSourcesWithProbes(
+ partitions: Int,
+ probe: TestSubscriber.Probe[(TopicPartition,
Source[ConsumerRecord[String, String], NotUsed])])
+ : Seq[(TopicPartition, TestSubscriber.Probe[ConsumerRecord[String,
String]])] =
+ probe
+ .expectNextN(partitions.toLong)
+ .map {
+ case (tp, subSource) =>
+ (tp, subSource.toMat(TestSink())(Keep.right).run())
+ }
+
+ def runForSubSource(
+ partition: Int,
+ subSourcesWithProbes: Seq[(TopicPartition,
TestSubscriber.Probe[ConsumerRecord[String, String]])])(
+ fun: TestSubscriber.Probe[ConsumerRecord[String, String]] => Unit)
=
+ subSourcesWithProbes
+ .find { case (tp, _) => tp.partition() == partition }
+ .foreach { case (_, probe) => fun(probe) }
+
+ val count = 20L
+ // de-coupling consecutive test runs with crossScalaVersions on build
+ val topicSuffix = Random.nextInt()
+ val topic1 = createTopic(topicSuffix, partitions = 2)
+ val group1 = createGroupId(1)
+ val tp0 = new TopicPartition(topic1, partition0)
+ val tp1 = new TopicPartition(topic1, partition1)
+ val consumerSettings = consumerDefaults
+ .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500") // 500
is the default value
+ .withProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
mode.assignor.getName)
+ .withGroupId(group1)
+
+ awaitProduce(produce(topic1, 0 to count.toInt, partition1))
+
+ PekkoConnectorsAssignor.clientIdToPartitionMap.set(
+ Map(
+ consumerClientId1 -> Set(tp0, tp1)))
+
+ log.debug("Subscribe to the topic (without demand)")
+ val probe1rebalanceActor = TestProbe()
+ val probe1subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe1rebalanceActor.ref)
+ val (control1, probe1) = Consumer
+
.plainPartitionedSource(consumerSettings.withClientId(consumerClientId1),
probe1subscription)
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ log.debug("Await initial partition assignment")
+ probe1rebalanceActor.expectMsg(
+ TopicPartitionsAssigned(probe1subscription,
+ Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
+
+ log.debug("read 2 sub sources returned by partitioned source")
+ probe1.request(2)
+ val probe1RunningSubSourceProbes = subSourcesWithProbes(partitions =
2, probe1)
+
+ log.debug("read one message from probe1 sub source for partition 1")
+ probe1RunningSubSourceProbes
+ .find { case (tp, _) => tp.partition() == partition1 }
+ .foreach { case (_, probe) => probe.requestNext() }
+
+ PekkoConnectorsAssignor.clientIdToPartitionMap.set(
+ Map(
+ consumerClientId1 -> Set(tp0),
+ consumerClientId2 -> Set(tp1)))
+
+ log.debug("Subscribe to the topic (without demand)")
+ val probe2rebalanceActor = TestProbe()
+ val probe2subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe2rebalanceActor.ref)
+ val (control2, probe2) = Consumer
+
.plainPartitionedSource(consumerSettings.withClientId(consumerClientId2),
probe2subscription)
+ .toMat(TestSink())(Keep.both)
+ .run()
+
+ probe2.request(1)
+ val probe2RunningSubSourceProbes = subSourcesWithProbes(partitions =
1, probe2)
+
+ log.debug("Await the rebalance to complete")
+ mode.expectSecondConsumerJoin(probe1rebalanceActor,
probe1subscription, probe2rebalanceActor,
+ probe2subscription, tp0, tp1)
+
+ log.debug("resume demand on both consumers")
+ runForSubSource(partition = 1,
probe1RunningSubSourceProbes)(_.request(count))
+ runForSubSource(partition = 1,
probe2RunningSubSourceProbes)(_.request(count))
+
+ log.debug("no further messages enqueued on probe1 as partition 1 is
balanced away")
+ runForSubSource(partition = 1,
probe1RunningSubSourceProbes)(_.expectComplete())
+
+ val probe2messages = probe2RunningSubSourceProbes
+ .find { case (tp, _) => tp.partition() == partition1 }
+ .toList
+ .flatMap {
+ case (_, probe) =>
+ probe.expectNextN(count)
}
- def runForSubSource(
- partition: Int,
- subSourcesWithProbes: Seq[(TopicPartition,
TestSubscriber.Probe[ConsumerRecord[String, String]])])(
- fun: TestSubscriber.Probe[ConsumerRecord[String, String]] => Unit) =
- subSourcesWithProbes
- .find { case (tp, _) => tp.partition() == partition }
- .foreach { case (_, probe) => fun(probe) }
-
- val count = 20L
- // de-coupling consecutive test runs with crossScalaVersions on build
- val topicSuffix = Random.nextInt()
- val topic1 = createTopic(topicSuffix, partitions = 2)
- val group1 = createGroupId(1)
- val tp0 = new TopicPartition(topic1, partition0)
- val tp1 = new TopicPartition(topic1, partition1)
- val consumerSettings = consumerDefaults
- .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500") // 500 is
the default value
- .withProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
classOf[PekkoConnectorsAssignor].getName)
- .withGroupId(group1)
-
- awaitProduce(produce(topic1, 0 to count.toInt, partition1))
-
- PekkoConnectorsAssignor.clientIdToPartitionMap.set(
- Map(
- consumerClientId1 -> Set(tp0, tp1)))
-
- log.debug("Subscribe to the topic (without demand)")
- val probe1rebalanceActor = TestProbe()
- val probe1subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe1rebalanceActor.ref)
- val (control1, probe1) = Consumer
-
.plainPartitionedSource(consumerSettings.withClientId(consumerClientId1),
probe1subscription)
- .toMat(TestSink.probe)(Keep.both)
- .run()
-
- log.debug("Await initial partition assignment")
- probe1rebalanceActor.expectMsg(
- TopicPartitionsAssigned(probe1subscription,
- Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
-
- log.debug("read 2 sub sources returned by partitioned source")
- probe1.request(2)
- val probe1RunningSubSourceProbes = subSourcesWithProbes(partitions = 2,
probe1)
-
- log.debug("read one message from probe1 sub source for partition 1")
- probe1RunningSubSourceProbes
- .find { case (tp, _) => tp.partition() == partition1 }
- .foreach { case (_, probe) => probe.requestNext() }
-
- PekkoConnectorsAssignor.clientIdToPartitionMap.set(
- Map(
- consumerClientId1 -> Set(tp0),
- consumerClientId2 -> Set(tp1)))
-
- log.debug("Subscribe to the topic (without demand)")
- val probe2rebalanceActor = TestProbe()
- val probe2subscription =
Subscriptions.topics(topic1).withRebalanceListener(probe2rebalanceActor.ref)
- val (control2, probe2) = Consumer
-
.plainPartitionedSource(consumerSettings.withClientId(consumerClientId2),
probe2subscription)
- .toMat(TestSink.probe)(Keep.both)
- .run()
-
- probe2.request(1)
- val probe2RunningSubSourceProbes = subSourcesWithProbes(partitions = 1,
probe2)
-
- log.debug("Await a revoke to consumer 1")
- probe1rebalanceActor.expectMsg(
- TopicPartitionsRevoked(probe1subscription,
- Set(new TopicPartition(topic1, partition0), new
TopicPartition(topic1, partition1))))
-
- log.debug("the rebalance finishes")
- probe1rebalanceActor.expectMsg(
- TopicPartitionsAssigned(probe1subscription, Set(new
TopicPartition(topic1, partition0))))
- probe2rebalanceActor.expectMsg(
- TopicPartitionsAssigned(probe2subscription, Set(new
TopicPartition(topic1, partition1))))
-
- log.debug("resume demand on both consumers")
- runForSubSource(partition = 1,
probe1RunningSubSourceProbes)(_.request(count))
- runForSubSource(partition = 1,
probe2RunningSubSourceProbes)(_.request(count))
-
- log.debug("no further messages enqueued on probe1 as partition 1 is
balanced away")
- runForSubSource(partition = 1,
probe1RunningSubSourceProbes)(_.expectComplete())
-
- val probe2messages = probe2RunningSubSourceProbes
- .find { case (tp, _) => tp.partition() == partition1 }
- .toList
- .flatMap {
- case (_, probe) =>
- probe.expectNextN(count)
- }
-
- probe2messages should have size count
+ probe2messages should have size count
- probe1.cancel()
- probe2.cancel()
+ probe1.cancel()
+ probe2.cancel()
- control1.isShutdown.futureValue shouldBe Done
- control2.isShutdown.futureValue shouldBe Done
+ control1.isShutdown.futureValue shouldBe Done
+ control2.isShutdown.futureValue shouldBe Done
+ }
}
}
}
@@ -290,3 +334,38 @@ class PekkoConnectorsAssignor extends
AbstractPartitionAssignor {
assignments.toMap.asJava
}
}
+
+/**
+ * Variant of [[PekkoConnectorsAssignor]] that uses the cooperative rebalance
protocol.
+ *
+ * The cooperative protocol requires that a partition never moves directly
from one member to
+ * another within a single rebalance: it must be absent from all assignments
for one generation
+ * (revoking it from its previous owner, which triggers a follow-up rebalance)
before it may be
+ * assigned to its new owner.
+ */
+class CooperativePekkoConnectorsAssignor extends PekkoConnectorsAssignor {
+
+ override def name(): String = "pekko-connector-kafka-test-cooperative"
+
+ override def supportedProtocols():
util.List[ConsumerPartitionAssignor.RebalanceProtocol] =
+ util.Arrays.asList(ConsumerPartitionAssignor.RebalanceProtocol.COOPERATIVE)
+
+ override def assign(
+ partitionsPerTopic: util.Map[String, Integer],
+ subscriptions: util.Map[String, ConsumerPartitionAssignor.Subscription])
+ : util.Map[String, util.List[TopicPartition]] = {
+ val desired = super.assign(partitionsPerTopic, subscriptions).asScala
+ val currentOwner: Map[TopicPartition, String] = (for {
+ (memberId, subscription) <- subscriptions.asScala.toSeq
+ tp <- subscription.ownedPartitions().asScala
+ } yield tp -> memberId).toMap
+
+ desired.map {
+ case (memberId, tps) =>
+ val withoutMovingPartitions = tps.asScala.filter { tp =>
+ currentOwner.get(tp).forall(_ == memberId)
+ }
+ memberId -> withoutMovingPartitions.asJava
+ }.asJava
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
