feat(example): concurrent processing using BatchMapper with BatchMapAsyncServer

[AdityaSoni19031997]

• Trying to add a minimal example to help end users so that they can leverage ConcurrentConsumption In UDFs when they're leveraging BatchMapper with BatchMapAsyncServer and ProcessPoolExecutor

[AdityaSoni19031997]

Ahh, I'm curious, What's the best way to test this out end-to-end on local? Is the expectation for end users is to create the NF local deployment via kind/minikube OR there's an alternative pattern followed?

Thanks!

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 92.99%. Comparing base (dc5092d) to head (b42f855).

Additional details and impacted files

@@           Coverage Diff           @@
##             main     #372   +/-   ##
=======================================
  Coverage   92.99%   92.99%           
=======================================
  Files          68       68           
  Lines        3656     3656           
  Branches      236      236           
=======================================
  Hits         3400     3400           
  Misses        190      190           
  Partials       66       66           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[yhl25]

Ahh, I'm curious, What's the best way to test this out end-to-end on local? Is the expectation for end users is to create the NF local deployment via kind/minikube OR there's an alternative pattern followed?

Thanks!

Yes, you will to test it using the kind cluster.

[AdityaSoni19031997]

Update: Testing is in progress, will share the SSs.
Thanks

[AdityaSoni19031997]

Hey Everyone! PFBA SS's from the local NF setup wrt latest commit sha available on OSS:

NB: I forwarded the local image to the cluster, didn't upload to any artifacts etc

POD LOGS:

More Procs (Over-subscribing)

[AdityaSoni19031997]

nit: Somewhat extensive logging is there on the pipeline for debugging, ideally can be skipped in real world scenarios.
Thanks!

[vigith]

can you make the linter pass?

[AdityaSoni19031997]

can you make the linter pass?

Yes, making a commit;
Thanks!

[vigith]

can you make the linter pass?

Still failing

2026-09-14T02:40:20.9342218Z ##[group]Run uv run ruff check .
2026-09-14T02:40:20.9342522Z �[36;1muv run ruff check .�[0m
2026-09-14T02:40:20.9364352Z shell: /usr/bin/bash -e {0}
2026-09-14T02:40:20.9364621Z env:
2026-09-14T02:40:20.9364866Z   UV_CACHE_DIR: /home/runner/work/_temp/setup-uv-cache
2026-09-14T02:40:20.9365188Z ##[endgroup]
2026-09-14T02:40:20.9764843Z UP035 `typing.Tuple` is deprecated, use `tuple` instead
2026-09-14T02:40:20.9765576Z   --> examples/batchmap/concurrent_sink/example.py:9:1
2026-09-14T02:40:20.9766092Z    |
2026-09-14T02:40:20.9766372Z  7 | from collections.abc import AsyncIterable
2026-09-14T02:40:20.9766734Z  8 | from concurrent.futures import ProcessPoolExecutor
2026-09-14T02:40:20.9767060Z  9 | from typing import Tuple, Optional
2026-09-14T02:40:20.9767325Z    | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-09-14T02:40:20.9767548Z 10 |
2026-09-14T02:40:20.9767760Z 11 | from pynumaflow.batchmapper import (
2026-09-14T02:40:20.9768013Z    |
2026-09-14T02:40:20.9768105Z 
2026-09-14T02:40:20.9768463Z UP045 [*] Use `X | None` for type annotations
2026-09-14T02:40:20.9768890Z   --> examples/batchmap/concurrent_sink/example.py:24:12
2026-09-14T02:40:20.9769181Z    |
2026-09-14T02:40:20.9769374Z 23 | # Process-level executor
2026-09-14T02:40:20.9769658Z 24 | _executor: Optional[ProcessPoolExecutor] = None
2026-09-14T02:40:20.9769967Z    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-09-14T02:40:20.9770208Z 25 |
2026-09-14T02:40:20.9770386Z 26 | gc.enable()
2026-09-14T02:40:20.9770570Z    |
2026-09-14T02:40:20.9770764Z help: Convert to `X | None`
2026-09-14T02:40:20.9770909Z 
2026-09-14T02:40:20.9771058Z UP006 [*] Use `tuple` instead of `Tuple` for type annotation
2026-09-14T02:40:20.9771418Z   --> examples/batchmap/concurrent_sink/example.py:54:16
2026-09-14T02:40:20.9771730Z    |
2026-09-14T02:40:20.9772002Z 53 | def _process_single_message_task(
2026-09-14T02:40:20.9772299Z 54 |     task_data: Tuple[str, list, bytes],
2026-09-14T02:40:20.9772747Z    |                ^^^^^
2026-09-14T02:40:20.9773077Z 55 | ) -> Tuple[str, list, bytes, bool]:
2026-09-14T02:40:20.9773611Z 56 |     """Worker function executed in ProcessPoolExecutor child process.
2026-09-14T02:40:20.9773958Z    |
2026-09-14T02:40:20.9774140Z help: Replace with `tuple`
2026-09-14T02:40:20.9774280Z 
2026-09-14T02:40:20.9774427Z UP006 [*] Use `tuple` instead of `Tuple` for type annotation
2026-09-14T02:40:20.9774779Z   --> examples/batchmap/concurrent_sink/example.py:55:6
2026-09-14T02:40:20.9775060Z    |
2026-09-14T02:40:20.9775247Z 53 | def _process_single_message_task(
2026-09-14T02:40:20.9775513Z 54 |     task_data: Tuple[str, list, bytes],
2026-09-14T02:40:20.9775781Z 55 | ) -> Tuple[str, list, bytes, bool]:
2026-09-14T02:40:20.9776020Z    |      ^^^^^
2026-09-14T02:40:20.9776300Z 56 |     """Worker function executed in ProcessPoolExecutor child process.
2026-09-14T02:40:20.9776662Z    |
2026-09-14T02:40:20.9776843Z help: Replace with `tuple`
2026-09-14T02:40:20.9776982Z 
2026-09-14T02:40:20.9777066Z Found 4 errors.
2026-09-14T02:40:20.9777277Z [*] 3 fixable with the `--fix` option.
2026-09-14T02:40:20.9792497Z ##[error]Process completed with exit code 1.

[AdityaSoni19031997]

I hope it should pass now, Sorry;
Thanks

Edits:

name: Lint

on:
  push:
    branches: [ "main", "release/*" ]
  pull_request:
    branches: [ "main", "release/*" ]

jobs:
  lint:
    name: Black & Ruff
    runs-on: ubuntu-latest
    defaults:
      run:
        working-directory: packages/pynumaflow/
    steps:
      - uses: actions/checkout@v3

      - name: Install uv
        uses: astral-sh/setup-uv@v4

      - name: Set up Python
        run: uv python install 3.10

      - name: Install dependencies
        run: uv sync --group dev

      - name: Black format check
        run: uv run black --check --verbose --exclude=pynumaflow/proto/ pynumaflow/ tests/ examples/

      - name: Ruff lint check
        run: uv run ruff check .

IMO: The WF shouldn't limit itself to Python 3.10 as it'll its official end-of-life (EOL) on October 31, 2026.
Thanks!

[vigith]

IMO: The WF shouldn't limit itself to Python 3.10 as it'll its official end-of-life (EOL) on October 31, 2026.

feel free to open a PR remove the support 🙂