fix(asyncio): Avoid InvalidStateError on late callbacks after cancel

.github/workflows/ci-build-release-wheels.yaml

@@ -23,6 +23,9 @@ on:
     tags:
       - '*'
 
+permissions:
+  contents: write
+
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
   cancel-in-progress: true
@@ -183,3 +186,28 @@ jobs:
         with:
           name: wheel-windows-py${{matrix.python.version}}
           path: dist/*.whl
+
+  publish-github-release:
+    name: Publish wheels to GitHub Release
+    needs:
+      - linux-wheel
+      - mac-wheels
+      - windows-wheels
+    runs-on: ubuntu-latest
+    steps:
+      - name: Download all wheel artifacts
+        uses: actions/download-artifact@v4
+        with:
+          path: dist
+          merge-multiple: true
+
+      - name: List wheels
+        run: find dist -name '*.whl' | sort
+
+      - name: Create GitHub Release with wheel assets
+        uses: softprops/action-gh-release@v2
+        with:
+          files: dist/**/*.whl
+          generate_release_notes: true
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

pulsar/__about__.py

@@ -16,4 +16,4 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-__version__='3.11.0a1'
+__version__='3.11.0a2'

pulsar/asyncio.py

@@ -828,6 +828,8 @@ async def close(self) -> None:
 
 def _set_future(future: asyncio.Future, result: _pulsar.Result, value: Any):
     def complete():
+        if future.done():
+            return
         if result == _pulsar.Result.Ok:
             future.set_result(value)
         else:

tests/asyncio_test.py

@@ -39,6 +39,7 @@
     Consumer,
     Producer,
     PulsarException,
+    _set_future,
 )
 from pulsar.schema import (  # pylint: disable=import-error
     AvroSchema,
@@ -484,5 +485,25 @@ class ExampleRecord(Record):  # pylint: disable=too-few-public-methods
         self.assertEqual(msg.value().int_field, 42)
 
 
+class AsyncioSetFutureTest(IsolatedAsyncioTestCase):
+    """Tests for asyncio bridge helpers (no live Pulsar broker)."""
+
+    async def test_set_future_noop_when_future_cancelled(self):
+        loop = asyncio.get_running_loop()
+        fut = loop.create_future()
+        fut.cancel()
+        _set_future(fut, _pulsar.Result.Ok, None)
+        await asyncio.sleep(0)
+        self.assertTrue(fut.cancelled())
+
+    async def test_set_future_noop_when_future_already_resolved(self):
+        loop = asyncio.get_running_loop()
+        fut = loop.create_future()
+        fut.set_result("first")
+        _set_future(fut, _pulsar.Result.Ok, "late")
+        await asyncio.sleep(0)
+        self.assertEqual(fut.result(), "first")
+
+
 if __name__ == '__main__':
     main()