fix: concurrency problems at local metadata store

.github/workflows/ci.yml

@@ -945,56 +945,107 @@ jobs:
           path: diagnostics*.gz
           retention-days: 1
 
- # FIXME: This is disabled until we fix the problem with the test when creating a new topic
- #  local_upgrade_test:
- #    name: Upgrade local cluster test on (${{ matrix.run }})
- #    needs: build_primary_binaries
- #    runs-on: ${{ matrix.os }}
- #    strategy:
- # #     fail-fast: false
- #      matrix:
- #        os: [ubuntu-latest]
- #        rust-target: [x86_64-unknown-linux-musl]
- #        run: [r1]
- #    steps:
- #      - uses: actions/checkout@v4
- #      # Download artifacts
- #      - name: Download artifact - fluvio
- #        uses: actions/download-artifact@v4
- #        with:
- #          name: fluvio-${{ matrix.rust-target }}
- #          path: ~/bin
- #      - name: Download artifact - fluvio-run
- #        uses: actions/download-artifact@v4
- #        with:
- #          name: fluvio-run-${{ matrix.rust-target }}
- #          path: ~/bin
- #      - name: Mark executable
- #        run: |
- #          chmod +x ~/bin/fluvio-run
- #          chmod +x ~/bin/fluvio && ~/bin/fluvio version
- #          echo "${HOME}/bin" >> $GITHUB_PATH
- #
- #      - name: Run upgrade test with CI artifacts
- #        timeout-minutes: 10
- #        env:
- #          TEST_DATA_BYTES: 10000
- #        run: |
- #          date
- #          make FLUVIO_MODE=local FLUVIO_BIN=~/bin/fluvio upgrade-test
- #
- #      - name: Run diagnostics
- #        if: ${{ !success() }}
- #        timeout-minutes: 5
- #        run: fluvio cluster diagnostics
- #      - name: Upload logs
- #        timeout-minutes: 5
- #        if: ${{ !success() }}
- #        uses: actions/upload-artifact@v4
- #        with:
- #          name: local_upgrade_${{ matrix.run }}_log
- #          path: diagnostics*.gz
- #          retention-days: 1
+  local_upgrade_test:
+    name: Upgrade local cluster test on (${{ matrix.run }})
+    needs: build_primary_binaries
+    runs-on: ${{ matrix.os }}
+    strategy:
+ #     fail-fast: false
+      matrix:
+        os: [ubuntu-latest]
+        rust-target: [x86_64-unknown-linux-musl]
+        run: [r1]
+    steps:
+      - uses: actions/checkout@v4
+      # Download artifacts
+      - name: Download artifact - fluvio
+        uses: actions/download-artifact@v4
+        with:
+          name: fluvio-${{ matrix.rust-target }}
+          path: ~/bin
+      - name: Download artifact - fluvio-run
+        uses: actions/download-artifact@v4
+        with:
+          name: fluvio-run-${{ matrix.rust-target }}
+          path: ~/bin
+      - name: Mark executable
+        run: |
+          chmod +x ~/bin/fluvio-run
+          chmod +x ~/bin/fluvio && ~/bin/fluvio version
+          echo "${HOME}/bin" >> $GITHUB_PATH
+
+      - name: Run upgrade test with CI artifacts
+        timeout-minutes: 10
+        env:
+          TEST_DATA_BYTES: 10000
+        run: |
+          date
+          make FLUVIO_MODE=local FLUVIO_BIN=~/bin/fluvio upgrade-test
+
+      - name: Run diagnostics
+        if: ${{ !success() }}
+        timeout-minutes: 5
+        run: fluvio cluster diagnostics
+      - name: Upload logs
+        timeout-minutes: 5
+        if: ${{ !success() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: local_upgrade_${{ matrix.run }}_log
+          path: diagnostics*.gz
+          retention-days: 1
+
+  local_resume_test:
+    name: Resume local cluster test on (${{ matrix.run }})
+    needs: build_primary_binaries
+    runs-on: ${{ matrix.os }}
+    strategy:
+ #     fail-fast: false
+      matrix:
+        os: [ubuntu-latest]
+        rust-target: [x86_64-unknown-linux-musl]
+        run: [r1]
+    steps:
+      - uses: actions/checkout@v4
+      # Download artifacts
+      - name: Download artifact - fluvio
+        uses: actions/download-artifact@v4
+        with:
+          name: fluvio-${{ matrix.rust-target }}
+          path: ~/bin
+      - name: Download artifact - fluvio-run
+        uses: actions/download-artifact@v4
+        with:
+          name: fluvio-run-${{ matrix.rust-target }}
+          path: ~/bin
+      - name: Mark executable
+        run: |
+          chmod +x ~/bin/fluvio-run
+          chmod +x ~/bin/fluvio && ~/bin/fluvio version
+          echo "${HOME}/bin" >> $GITHUB_PATH
+      - name: Install Parallel
+        run: sudo apt-get install -y parallel
+
+      - name: Run resume test with CI artifacts
+        timeout-minutes: 10
+        env:
+          TEST_DATA_BYTES: 10000
+        run: |
+          date
+          make FLUVIO_BIN=~/bin/fluvio resume-test
+
+      - name: Run diagnostics
+        if: ${{ !success() }}
+        timeout-minutes: 5
+        run: fluvio cluster diagnostics
+      - name: Upload logs
+        timeout-minutes: 5
+        if: ${{ !success() }}
+        uses: actions/upload-artifact@v4
+        with:
+          name: local_upgrade_${{ matrix.run }}_log
+          path: diagnostics*.gz
+          retention-days: 1
 
   # Smoke test across different version of fluvio
   cli_smoke:

Cargo.toml

@@ -115,6 +115,7 @@ mimalloc = "0.1.39"
 mime = "0.3"
 nix = { version = "0.29.0", default-features = false }
 once_cell = "1.7.2"
+parking_lot = { version = "0.12.3", default-features = false }
 pin-project = "1.1.0"
 portpicker = "0.1.1"
 proc-macro2 = "1.0"

crates/fluvio-stream-dispatcher/Cargo.toml

@@ -26,6 +26,7 @@ once_cell = { workspace = true }
 serde = { workspace = true,  features = ['derive'] }
 serde_json = { workspace = true, optional = true }
 serde_yaml = { workspace = true, optional = true }
+parking_lot = { workspace = true, features = ["send_guard"] }
 tokio = { workspace = true, features = ["macros"] }
 tracing = { workspace = true }
 tempfile = { workspace = true }