feat: transactions

.github/workflows/rust.yml

@@ -30,7 +30,22 @@ jobs:
  pulsar-version: [ 2.10.4, 2.11.2, 3.0.4, 3.1.3 ]
  steps:
  - name: Start Pulsar Standalone Container
- run: docker run --name pulsar -p 6650:6650 -p 8080:8080 -d -e GITHUB_ACTIONS=true -e CI=true apachepulsar/pulsar:${{ matrix.pulsar-version }} bin/pulsar standalone
+ run: |
+ docker run --name pulsar \
+ -p 6650:6650 \
+ -p 8080:8080 \
+ -d \
+ -e GITHUB_ACTIONS=true \
+ -e CI=true \
+ -e PULSAR_PREFIX_transactionCoordinatorEnabled="true" \
+ -e PULSAR_PREFIX_systemTopicEnabled="true" \
+ -e PULSAR_PREFIX_metadataStoreUrl="zk:127.0.0.1:2181" \
+ apachepulsar/pulsar:${{ matrix.pulsar-version }} sh -c \
+ "bin/apply-config-from-env.py \
+ conf/standalone.conf && \
+ bin/pulsar-daemon start zookeeper && \
+ bin/pulsar initialize-transaction-coordinator-metadata -cs 127.0.0.1:2181 -c standalone && \
+ bin/pulsar standalone"
  - uses: actions/checkout@v3
  - uses: Swatinem/rust-cache@v2
  - name: Run tests

examples/transaction.rs

@@ -0,0 +1,191 @@
+#[macro_use]
+extern crate serde;
+use std::{env, time::Duration};
+
+use futures::TryStreamExt;
+use pulsar::{
+ authentication::oauth2::OAuth2Authentication, message::proto, producer, Authentication,
+ Consumer, DeserializeMessage, Error as PulsarError, Payload, Pulsar, SerializeMessage,
+ TokioExecutor,
+};
+
+#[derive(Serialize, Deserialize, Debug)]
+struct TestData {
+ data: String,
+}
+
+impl SerializeMessage for TestData {
+ fn serialize_message(input: Self) -> Result<producer::Message, PulsarError> {
+ let payload = serde_json::to_vec(&input).map_err(|e| PulsarError::Custom(e.to_string()))?;
+ Ok(producer::Message {
+ payload,
+ ..Default::default()
+ })
+ }
+}
+
+impl DeserializeMessage for TestData {
+ type Output = Result<TestData, serde_json::Error>;
+
+ fn deserialize_message(payload: &Payload) -> Self::Output {
+ serde_json::from_slice(&payload.data)
+ }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), pulsar::Error> {
+ env_logger::init();
+
+ let addr = env::var("PULSAR_ADDRESS")
+ .ok()
+ .unwrap_or_else(|| "pulsar://127.0.0.1:6650".to_string());
+
+ let mut builder = Pulsar::builder(addr, TokioExecutor).with_transactions();
+
+ if let Ok(token) = env::var("PULSAR_TOKEN") {
+ let authentication = Authentication {
+ name: "token".to_string(),
+ data: token.into_bytes(),
+ };
+
+ builder = builder.with_auth(authentication);
+ } else if let Ok(oauth2_cfg) = env::var("PULSAR_OAUTH2") {
+ builder = builder.with_auth_provider(OAuth2Authentication::client_credentials(
+ serde_json::from_str(oauth2_cfg.as_str())
+ .unwrap_or_else(|_| panic!("invalid oauth2 config [{}]", oauth2_cfg.as_str())),
+ ));
+ }
+
+ let pulsar: Pulsar<_> = builder.build().await?;
+
+ let input_topic = "persistent://public/default/test-input-topic";
+ let output_topic_one = "persistent://public/default/test-output-topic-1";
+ let output_topic_two = "persistent://public/default/test-output-topic-2";
+
+ let producer_builder = pulsar.producer().with_options(producer::ProducerOptions {
+ schema: Some(proto::Schema {
+ r#type: proto::schema::Type::String as i32,
+ ..Default::default()
+ }),
+ ..Default::default()
+ });
+
+ let mut input_producer = producer_builder
+ .clone()
+ .with_topic(input_topic)
+ .build()
+ .await?;
+
+ let mut output_producer_one = producer_builder
+ .clone()
+ .with_topic(output_topic_one)
+ .build()
+ .await?;
+
+ let mut output_producer_two = producer_builder
+ .clone()
+ .with_topic(output_topic_two)
+ .build()
+ .await?;
+
+ let mut input_consumer: Consumer<TestData, _> = pulsar
+ .consumer()
+ .with_topic(input_topic)
+ .with_subscription("test-input-subscription")
+ .build()
+ .await?;
+
+ let mut output_consumer_one: Consumer<TestData, _> = pulsar
+ .consumer()
+ .with_topic(output_topic_one)
+ .with_subscription("test-output-subscription-1")
+ .build()
+ .await?;
+
+ let mut output_consumer_two: Consumer<TestData, _> = pulsar
+ .consumer()
+ .with_topic(output_topic_two)
+ .with_subscription("test-output-subscription-2")
+ .build()
+ .await?;
+
+ let count = 2;
+
+ for i in 0..count {
+ input_producer
+ .send_non_blocking(TestData {
+ data: format!("Hello Pulsar! count : {}", i),
+ })
+ .await?
+ .await?;
+ }
+
+ for i in 0..count {
+ let msg = input_consumer
+ .try_next()
+ .await?
+ .expect("No message received");
+
+ let txn = pulsar
+ .new_txn()?
+ .with_timeout(Duration::from_secs(10))
+ .build()
+ .await?;
+
+ output_producer_one
+ .create_message()
+ .with_content(TestData {
+ data: format!("Hello Pulsar! output_topic_one count : {}", i),
+ })
+ .with_txn(&txn)
+ .send_non_blocking()
+ .await?;
+
+ output_producer_two
+ .create_message()
+ .with_content(TestData {
+ data: format!("Hello Pulsar! output_topic_two count : {}", i),
+ })
+ .with_txn(&txn)
+ .send_non_blocking()
+ .await?;
+
+ input_consumer.txn_ack(&msg, &txn).await?;
+
+ if let Err(e) = txn.commit().await {
+ match e {
+ pulsar::Error::Transaction(pulsar::error::TransactionError::Conflict) => {
+ // If TransactionConflictException is not thrown,
+ // you need to redeliver or negativeAcknowledge this message,
+ // or else this message will not be received again.
+ input_consumer.nack(&msg).await?;
+ }
+ _ => (),
+ }
+
+ txn.abort().await?;
+
+ return Err(e);
+ }
+ }
+
+ for _ in 0..count {
+ let msg = output_consumer_one
+ .try_next()
+ .await?
+ .expect("No message received");
+
+ println!("Received transaction message: {:?}", msg.deserialize());
+ }
+
+ for _ in 0..count {
+ let msg = output_consumer_two
+ .try_next()
+ .await?
+ .expect("No message received");
+
+ println!("Received transaction message: {:?}", msg.deserialize());
+ }
+
+ Ok(())
+}

src/client.rs

@@ -21,6 +21,7 @@ use crate::{
  },
  producer::{self, ProducerBuilder, SendFuture},
  service_discovery::ServiceDiscovery,
+ transaction::{coord::TransactionCoordinatorClient, TransactionBuilder},
 };
 
 /// Helper trait for consumer deserialization
@@ -156,6 +157,8 @@ impl<'a> SerializeMessage for &'a str {
 pub struct Pulsar<Exe: Executor> {
  pub(crate) manager: Arc<ConnectionManager<Exe>>,
  service_discovery: Arc<ServiceDiscovery<Exe>>,
+ /// The transaction coordinator client, if transactions are enabled.
+ tc_client: Option<Arc<TransactionCoordinatorClient<Exe>>>,
  // this field is an Option to avoid a cyclic dependency between Pulsar
  // and run_producer: the run_producer loop needs a client to create
  // a multitopic producer, this producer stores internally a copy
@@ -179,6 +182,7 @@ impl<Exe: Executor> Pulsar<Exe> {
  connection_retry_parameters: Option<ConnectionRetryOptions>,
  operation_retry_parameters: Option<OperationRetryOptions>,
  tls_options: Option<TlsOptions>,
+ transactions_enabled: bool,
  outbound_channel_size: Option<usize>,
  executor: Exe,
  ) -> Result<Self, Error> {
@@ -221,17 +225,24 @@ impl<Exe: Executor> Pulsar<Exe> {
  let (producer, producer_rx) = mpsc::unbounded();
 
  let mut client = Pulsar {
- manager,
+ manager: Arc::clone(&manager),
  service_discovery,
+ tc_client: None,
  producer: None,
  operation_retry_options,
  executor,
  };
 
+ if transactions_enabled {
+ let tc_client = TransactionCoordinatorClient::new(client.clone(), manager).await?;
+ client.tc_client = Some(Arc::new(tc_client));
+ }
+
  let _ = client
  .executor
  .spawn(Box::pin(run_producer(client.clone(), producer_rx)));
  client.producer = Some(producer);
+
  Ok(client)
  }
 
@@ -256,10 +267,32 @@ impl<Exe: Executor> Pulsar<Exe> {
  operation_retry_options: None,
  tls_options: None,
  outbound_channel_size: None,
+ transactions_enabled: false,
  executor,
  }
  }
 
+ /// Creates a new transaction builder. If transactions were not enabled when creating the
+ /// Pulsar client, this will return an error.
+ ///
+ /// ```rust,no_run
+ /// use pulsar::Transaction;
+ ///
+ /// # async fn run(pulsar: pulsar::Pulsar<pulsar::TokioExecutor>) -> Result<(), pulsar::Error> {
+ /// let txn = pulsar.new_txn().with_timeout(1000).build().await?;
+ /// # Ok(())
+ /// # }
+ pub fn new_txn(&self) -> Result<TransactionBuilder<Exe>, Error> {
+ if let Some(tc_client) = self.tc_client.as_ref() {
+ Ok(TransactionBuilder::new(
+ Arc::clone(&self.executor),
+ Arc::clone(tc_client),
+ ))
+ } else {
+ Err(Error::Custom("Transactions are not enabled".into()))
+ }
+ }
+
  /// creates a consumer builder
  ///
  /// ```rust,no_run
@@ -457,6 +490,7 @@ pub struct PulsarBuilder<Exe: Executor> {
  operation_retry_options: Option<OperationRetryOptions>,
  tls_options: Option<TlsOptions>,
  outbound_channel_size: Option<usize>,
+ transactions_enabled: bool,
  executor: Exe,
 }
 
@@ -560,6 +594,13 @@ impl<Exe: Executor> PulsarBuilder<Exe> {
  self
  }
 
+ /// Enable transactions on this Pulsar client
+ #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+ pub fn with_transactions(mut self) -> Self {
+ self.transactions_enabled = true;
+ self
+ }
+
  /// creates the Pulsar client and connects it
  #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
  pub async fn build(self) -> Result<Pulsar<Exe>, Error> {
@@ -570,6 +611,7 @@ impl<Exe: Executor> PulsarBuilder<Exe> {
  operation_retry_options,
  tls_options,
  outbound_channel_size,
+ transactions_enabled,
  executor,
  } = self;
 
@@ -579,6 +621,7 @@ impl<Exe: Executor> PulsarBuilder<Exe> {
  connection_retry_options,
  operation_retry_options,
  tls_options,
+ transactions_enabled,
  outbound_channel_size,
  executor,
  )