chore: fluvio update to 0.11.11 (#445)

* chore: ci, change latest fluvio workflow to python 3.12

* chore: Cargo.toml sort deps

* chore: update fluvio deps

* fix: flvuio client code updates

.github/workflows/latest-dev-fluvio.yaml

@@ -15,7 +15,7 @@ jobs:
       matrix:
         os: [ubuntu-latest]
         rust: [stable]
-        python-version: ["3.11"]
+        python-version: ["3.12"]
     steps:
       - uses: actions/checkout@v4
       - name: Install rust ${{ matrix.rust }}

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "_fluvio_python"
-version = "0.16.2"
+version = "0.16.3"
 edition = "2021"
 authors = ["Fluvio Contributors <[email protected]>"]
 
@@ -14,30 +14,30 @@ log = "^0.4.17"
 
 [dependencies]
 anyhow = "1.0"
+async-h1 = "2.3.3"
+async-lock = "3.3.0"
+async-std = "1.6.5"
 cc = "=1.1.5"
+dirs = "5.0.1"
+futures = "0.3.30"
+hex = "0.4.2"
+http-types = "2.6.0"
+md-5 = "0.10.0"
 pyo3 = { version = "0.20.2", features = ["extension-module"] }
 pyo3-asyncio = { version = "0.20.0", features = ["testing", "attributes", "async-std-runtime"] }
-thiserror = "1.0.21"
+rpassword = "7.3.1"
 serde = "1.0.117"
+serde_json = "1.0.59"
+serde_urlencoded = "0.7.1"
+thiserror = "1.0.21"
+tokio = { version = "1.36.0", default-features = false, features = ["macros"] }
+toml = "0.8.1"
 tracing = "0.1.37"
 url = "2.5.0"
-dirs = "5.0.1"
-http-types = "2.6.0"
-tokio = { version = "1.36.0", default-features = false, features = ["macros"] }
-async-h1 = "2.3.3"
-async-std = "1.6.5"
-serde_urlencoded = "0.7.1"
-md-5 = "0.10.0"
-hex = "0.4.2"
 webbrowser = "1.0.0"
-toml = "0.8.1"
-serde_json = "1.0.59"
-rpassword = "7.3.1"
-futures = "0.3.30"
-async-lock = "3.3.0"
 
-fluvio-future = { version = "0.6.2", features = ["task", "io", "native2_tls", "subscriber"] }
-fluvio = { features = ["admin", "rustls"], git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.9" }
-fluvio-types = { git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.9" }
-fluvio-sc-schema = { git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.9" }
-fluvio-controlplane-metadata = { git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.9" }
+fluvio-future = { version = "0.7.0", features = ["task", "io", "native_tls", "subscriber"] }
+fluvio = { features = ["admin", "rustls"], git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.11" }
+fluvio-types = { git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.11" }
+fluvio-sc-schema = { git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.11" }
+fluvio-controlplane-metadata = { git = "https://github.com/infinyon/fluvio.git", tag = "v0.11.11" }

integration-tests/smartmodule-filter-on-a/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2021"
 crate-type = ['cdylib']
 
 [dependencies]
-fluvio-smartmodule = "0.7.2"
+fluvio-smartmodule = "0.7.4"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 

integration-tests/test_fluvio_python.py

@@ -796,4 +796,5 @@ def test_admin_paritions(self):
 
         # list partitions
         all_partitions = fluvio_admin.list_partitions([])
+        print(all_partitions)
         self.assertNotEqual(len(all_partitions), 0)

src/lib.rs

@@ -16,7 +16,7 @@ use fluvio::{
     MultiplePartitionConsumer as NativeMultiplePartitionConsumer, Offset as NativeOffset,
     PartitionConsumer as NativePartitionConsumer,
     PartitionSelectionStrategy as NativePartitionSelectionStrategy,
-    TopicProducer as NativeTopicProducer,
+    TopicProducerPool as NativeTopicProducer,
 };
 use fluvio_controlplane_metadata::message::{Message as NativeMessage, MsgType as NativeMsgType};
 use fluvio_controlplane_metadata::partition::PartitionSpec as NativePartitionSpec;
@@ -148,9 +148,10 @@ impl Fluvio {
     }
 
     fn topic_producer(&self, topic: String, py: Python) -> PyResult<TopicProducer> {
-        Ok(TopicProducer(py.allow_threads(move || {
+        let native_prod = py.allow_threads(move || {
             run_block_on(self.0.topic_producer(topic)).map_err(error_to_py_err)
-        })?))
+        })?;
+        Ok(TopicProducer(Arc::new(native_prod)))
     }
 }
 
@@ -699,7 +700,7 @@ impl RecordMetadata {
 
 #[derive(Clone)]
 #[pyclass]
-struct TopicProducer(NativeTopicProducer);
+struct TopicProducer(Arc<NativeTopicProducer>);
 
 #[pymethods]
 impl TopicProducer {