tx: reject producers with timeouts larger than transaction_max_timeou…

…t_ms error code is inline with Apache Kafka return code invalid_transaction_timeout(50)
redpanda-data · Jul 18, 2024 · a9cdabb · a9cdabb
1 parent 2d7f722
commit a9cdabb
Show file tree

Hide file tree

Showing 7 changed files with 25 additions and 5 deletions.
diff --git a/src/v/cluster/tx_errc.cc b/src/v/cluster/tx_errc.cc
@@ -70,6 +70,8 @@ std::ostream& operator<<(std::ostream& o, errc err) {
         return o << "tx::errc::partition_disabled";
     case errc::concurrent_transactions:
         return o << "tx::errc::concurrent_transactions";
+    case errc::invalid_timeout:
+        return o << "tx::errc::invalid_timeout";
     }
     return o;
 }

diff --git a/src/v/cluster/tx_errc.h b/src/v/cluster/tx_errc.h
@@ -48,6 +48,8 @@ enum class errc {
     tx_id_not_found,
     partition_disabled,
     concurrent_transactions,
+    // invalid timeout requested by the client.
+    invalid_timeout,
 };
 
 std::ostream& operator<<(std::ostream& o, errc err);

diff --git a/src/v/cluster/tx_gateway_frontend.cc b/src/v/cluster/tx_gateway_frontend.cc
@@ -708,6 +708,19 @@ ss::future<cluster::init_tm_tx_reply> tx_gateway_frontend::init_tm_tx_locally(
       tx_id,
       transaction_timeout_ms);
 
+    if (unlikely(
+          transaction_timeout_ms
+          > config::shard_local_cfg().transaction_max_timeout_ms())) {
+        vlog(
+          txlog.warn,
+          "[tx_id={}] Transactional timeout requested {}ms exceeds configured "
+          "maximum timeout {}ms",
+          tx_id,
+          transaction_timeout_ms,
+          config::shard_local_cfg().transaction_max_timeout_ms());
+        co_return init_tm_tx_reply{tx::errc::invalid_timeout};
+    }
+
     model::ntp tx_ntp(model::tx_manager_nt.ns, model::tx_manager_nt.tp, tm);
     auto shard = _shard_table.local().shard_for(tx_ntp);
 

diff --git a/src/v/kafka/server/errors.h b/src/v/kafka/server/errors.h
@@ -161,6 +161,8 @@ constexpr error_code map_tx_errc(cluster::tx::errc ec) {
     case cluster::tx::errc::request_rejected:
     case cluster::tx::errc::unknown_server_error:
         return error_code::unknown_server_error;
+    case cluster::tx::errc::invalid_timeout:
+        return error_code::invalid_transaction_timeout;
     }
 }
 

diff --git a/tests/java/verifiers/src/main/java/io/vectorized/tx_verifier/Verifier.java b/tests/java/verifiers/src/main/java/io/vectorized/tx_verifier/Verifier.java
@@ -15,6 +15,7 @@ public static interface StringAction {
   }
 
   final static int RETRY_TIMEOUT_SEC = 5;
+  final static int MAX_TRANSACTION_TIMEOUT_MS = 900000;
   final static String txId1 = "tx1";
   final static String txId2 = "tx2";
   final static String topic1 = "topic1";
@@ -407,7 +408,7 @@ static void readCommittedSeekRespectsLongHangingTx(String connection)
     TxProducer producer = null;
     TxConsumer consumer = null;
     try {
-      producer = new TxProducer(connection, txId1, Integer.MAX_VALUE);
+      producer = new TxProducer(connection, txId1, MAX_TRANSACTION_TIMEOUT_MS);
       producer.initTransactions();
       producer.beginTransaction();
       long offset = producer.send(topic1, "key1", "value1");

diff --git a/tests/rptest/transactions/transactions_test.py b/tests/rptest/transactions/transactions_test.py
@@ -391,7 +391,7 @@ def transaction_id_expiration_test(self):
         producer = ck.Producer({
             'bootstrap.servers': self.redpanda.brokers(),
             'transactional.id': '0',
-            'transaction.timeout.ms': 3600000,  # to avoid timing out
+            'transaction.timeout.ms': 900000,  # to avoid timing out
         })
         producer.init_transactions()
         producer.begin_transaction()
@@ -574,7 +574,7 @@ def graceful_leadership_transfer_tx_coordinator_test(self):
             ck.Producer({
                 'bootstrap.servers': self.redpanda.brokers(),
                 'transactional.id': str(i),
-                'transaction.timeout.ms': 1000000,
+                'transaction.timeout.ms': 900000,
             }) for i in range(0, p_count)
         ]
 

diff --git a/tests/rptest/transactions/tx_admin_api_test.py b/tests/rptest/transactions/tx_admin_api_test.py
@@ -93,12 +93,12 @@ def test_mark_transaction_expired(self):
         producer1 = ck.Producer({
             'bootstrap.servers': self.redpanda.brokers(),
             'transactional.id': '0',
-            'transaction.timeout.ms': 3600000,  # avoid auto timeout
+            'transaction.timeout.ms': 900000,  # avoid auto timeout
         })
         producer2 = ck.Producer({
             'bootstrap.servers': self.redpanda.brokers(),
             'transactional.id': '1',
-            'transaction.timeout.ms': 3600000,  # avoid auto timeout
+            'transaction.timeout.ms': 900000,  # avoid auto timeout
         })
         producer1.init_transactions()
         producer2.init_transactions()