transactions

.env

@@ -38,8 +38,8 @@ AWS_ALLOW_HTTP="true"
 # minio storage engine
 # create a bucket "tansu" at: http://localhost:9001/buckets
 # or: mc mb local/tansu
-STORAGE_ENGINE="minio=s3://tansu/"
+# STORAGE_ENGINE="minio=s3://tansu/"
 
 # PostgreSQL storage engine
 # STORAGE_ENGINE="pg=postgres://postgres:postgres@db"
-# STORAGE_ENGINE="pg=postgres://postgres:postgres@localhost"
+STORAGE_ENGINE="pg=postgres://postgres:postgres@localhost"

.github/workflows/ci.yml

@@ -17,8 +17,13 @@ jobs:
       matrix:
         os:
           - ubuntu-latest
+        pg:
+          - postgres:16
     steps:
       - uses: actions/checkout@v4
+      - run: docker compose up --detach db
+        env:
+          POSTGRES_IMAGE: ${{matrix.pg}}
       - uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
           components: clippy, rustfmt

compose.yaml

@@ -1,7 +1,7 @@
 ---
 services:
   db:
-    image: postgres:16.4
+    image: ${POSTGRES_IMAGE:-postgres:16.4}
     environment:
       PGUSER: postgres
       POSTGRES_PASSWORD: postgres

justfile

@@ -9,7 +9,7 @@ test:
     cargo test --workspace --all-targets
 
 clippy:
-    cargo clippy --all-targets
+    cargo clippy --all-targets -- -D warnings
 
 fmt:
     cargo fmt --all
@@ -118,7 +118,7 @@ tansu-1:
         --work-dir work-dir/tansu-1 2>&1 | tee tansu.log
 
 kafka-proxy:
-    docker run -d -p 19092:9092 apache/kafka:3.8.0
+    docker run -d -p 19092:9092 apache/kafka:3.9.0
 
 proxy:
     ./target/debug/tansu-proxy 2>&1 | tee proxy.log

tansu-kafka-sans-io/build.rs

@@ -1127,7 +1127,7 @@ fn all(pattern: &str) -> Result<Vec<Message>> {
         .and_then(|mut paths| {
             paths.try_fold(Vec::new(), |mut acc, p| {
                 p.map_err(Into::into)
-                    // .inspect(|path| eprintln!("path: {path:?}"))
+                    .inspect(|path| println!("cargo::rerun-if-changed={}", path.display()))
                     .and_then(read_value)
                     .and_then(|v| Message::try_from(&Wv::from(&v)).map_err(Into::into))
                     .map(|m| {

tansu-kafka-sans-io/message/AddOffsetsToTxnRequest.json

@@ -23,7 +23,9 @@
   // Version 2 adds the support for new error code PRODUCER_FENCED.
   //
   // Version 3 enables flexible versions.
-  "validVersions": "0-3",
+  //
+  // Version 4 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "TransactionalId", "type": "string", "versions": "0+", "entityType": "transactionalId",

tansu-kafka-sans-io/message/AddOffsetsToTxnResponse.json

@@ -22,7 +22,9 @@
   // Version 2 adds the support for new error code PRODUCER_FENCED.
   //
   // Version 3 enables flexible versions.
-  "validVersions": "0-3",
+  //
+  // Version 4 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

tansu-kafka-sans-io/message/AddPartitionsToTxnRequest.json

@@ -25,9 +25,11 @@
   // Version 3 enables flexible versions.
   //
   // Version 4 adds VerifyOnly field to check if partitions are already in transaction and adds support to batch multiple transactions.
+  //
+  // Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
   // Versions 3 and below will be exclusively used by clients and versions 4 and above will be used by brokers.
   "latestVersionUnstable": false,
-  "validVersions": "0-4",
+  "validVersions": "0-5",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "Transactions", "type": "[]AddPartitionsToTxnTransaction", "versions":  "4+",

tansu-kafka-sans-io/message/AddPartitionsToTxnResponse.json

@@ -24,7 +24,9 @@
   // Version 3 enables flexible versions.
   //
   // Version 4 adds support to batch multiple transactions and a top level error code.
-  "validVersions": "0-4",
+  //
+  // Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-5",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

tansu-kafka-sans-io/message/ConsumerGroupDescribeRequest.json

@@ -19,10 +19,6 @@
   "listeners": ["zkBroker", "broker"],
   "name": "ConsumerGroupDescribeRequest",
   "validVersions": "0",
-  // The ConsumerGroupDescribe API is added as part of KIP-848 and is still
-  // under development. Hence, the API is not exposed by default by brokers
-  // unless explicitly enabled.
-  "latestVersionUnstable": true,
   "flexibleVersions": "0+",
   "fields": [
     { "name": "GroupIds", "type": "[]string", "versions": "0+", "entityType": "groupId",

tansu-kafka-sans-io/message/ConsumerGroupDescribeResponse.json

@@ -87,13 +87,7 @@
     ]},
     { "name": "Assignment", "versions": "0+", "fields": [
       { "name": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+",
-        "about": "The assigned topic-partitions to the member." },
-      { "name": "Error", "type": "int8", "versions": "0+",
-        "about": "The assigned error." },
-      { "name": "MetadataVersion", "type": "int32", "versions": "0+",
-        "about": "The assignor metadata version." },
-      { "name": "MetadataBytes", "type": "bytes", "versions": "0+",
-        "about": "The assignor metadata bytes." }
+        "about": "The assigned topic-partitions to the member." }
     ]}
   ]
 }

tansu-kafka-sans-io/message/DescribeTopicPartitionsRequest.json

@@ -0,0 +1,40 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 75,
+  "type": "request",
+  "listeners": ["broker"],
+  "name": "DescribeTopicPartitionsRequest",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "Topics", "type": "[]TopicRequest", "versions": "0+",
+      "about": "The topics to fetch details for.",
+      "fields": [
+        { "name": "Name", "type": "string", "versions": "0+",
+          "about": "The topic name", "versions": "0+", "entityType": "topicName"}
+      ]
+    },
+    { "name": "ResponsePartitionLimit", "type": "int32", "versions": "0+", "default": "2000",
+      "about": "The maximum number of partitions included in the response." },
+    { "name": "Cursor", "type": "Cursor", "versions": "0+", "nullableVersions": "0+", "default": "null",
+      "about": "The first topic and partition index to fetch details for.", "fields": [
+      { "name": "TopicName", "type": "string", "versions": "0+",
+        "about": "The name for the first topic to process", "versions": "0+", "entityType": "topicName"},
+      { "name": "PartitionIndex", "type": "int32", "versions": "0+", "about": "The partition index to start with"}
+    ]}
+  ]
+}

tansu-kafka-sans-io/message/DescribeTopicPartitionsResponse.json

@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 75,
+  "type": "response",
+  "name": "DescribeTopicPartitionsResponse",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
+      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+    { "name": "Topics", "type": "[]DescribeTopicPartitionsResponseTopic", "versions": "0+",
+      "about": "Each topic in the response.", "fields": [
+      { "name": "ErrorCode", "type": "int16", "versions": "0+",
+        "about": "The topic error, or 0 if there was no error." },
+      { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName", "nullableVersions": "0+",
+        "about": "The topic name." },
+      { "name": "TopicId", "type": "uuid", "versions": "0+", "ignorable": true, "about": "The topic id." },
+      { "name": "IsInternal", "type": "bool", "versions": "0+", "default": "false", "ignorable": true,
+        "about": "True if the topic is internal." },
+      { "name": "Partitions", "type": "[]DescribeTopicPartitionsResponsePartition", "versions": "0+",
+        "about": "Each partition in the topic.", "fields": [
+        { "name": "ErrorCode", "type": "int16", "versions": "0+",
+          "about": "The partition error, or 0 if there was no error." },
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index." },
+        { "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
+          "about": "The ID of the leader broker." },
+        { "name": "LeaderEpoch", "type": "int32", "versions": "0+", "default": "-1", "ignorable": true,
+          "about": "The leader epoch of this partition." },
+        { "name": "ReplicaNodes", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
+          "about": "The set of all nodes that host this partition." },
+        { "name": "IsrNodes", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
+          "about": "The set of nodes that are in sync with the leader for this partition." },
+        { "name": "EligibleLeaderReplicas", "type": "[]int32", "default": "null", "entityType": "brokerId",
+          "versions": "0+", "nullableVersions": "0+",
+          "about": "The new eligible leader replicas otherwise." },
+        { "name": "LastKnownElr", "type": "[]int32", "default": "null", "entityType": "brokerId",
+          "versions": "0+", "nullableVersions": "0+",
+          "about": "The last known ELR." },
+        { "name": "OfflineReplicas", "type": "[]int32", "versions": "0+", "ignorable": true, "entityType": "brokerId",
+          "about": "The set of offline replicas of this partition." }]},
+      { "name": "TopicAuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
+        "about": "32-bit bitfield to represent authorized operations for this topic." }]
+    },
+    { "name": "NextCursor", "type": "Cursor", "versions": "0+", "nullableVersions": "0+", "default": "null",
+      "about": "The next topic and partition index to fetch details for.", "fields": [
+      { "name": "TopicName", "type": "string", "versions": "0+",
+        "about": "The name for the first topic to process", "versions": "0+", "entityType": "topicName"},
+      { "name": "PartitionIndex", "type": "int32", "versions": "0+", "about": "The partition index to start with"}
+    ]}
+  ]
+}

tansu-kafka-sans-io/message/EndTxnRequest.json

@@ -23,7 +23,9 @@
   // Version 2 adds the support for new error code PRODUCER_FENCED.
   //
   // Version 3 enables flexible versions.
-  "validVersions": "0-3",
+  //
+  // Version 4 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "TransactionalId", "type": "string", "versions": "0+", "entityType": "transactionalId",

tansu-kafka-sans-io/message/EndTxnResponse.json

@@ -22,7 +22,9 @@
   // Version 2 adds the support for new error code PRODUCER_FENCED.
   //
   // Version 3 enables flexible versions.
-  "validVersions": "0-3",
+  //
+  // Version 4 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-4",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

tansu-kafka-sans-io/message/FindCoordinatorRequest.json

@@ -25,7 +25,9 @@
   // Version 3 is the first flexible version.
   //
   // Version 4 adds support for batching via CoordinatorKeys (KIP-699)
-  "validVersions": "0-4",
+  //
+  // Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-5",
   "deprecatedVersions": "0",
   "flexibleVersions": "3+",
   "fields": [

tansu-kafka-sans-io/message/FindCoordinatorResponse.json

@@ -24,7 +24,9 @@
   // Version 3 is the first flexible version.
   //
   // Version 4 adds support for batching via Coordinators (KIP-699)
-  "validVersions": "0-4",
+  //
+  // Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-5",
   "flexibleVersions": "3+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "1+", "ignorable": true,

tansu-kafka-sans-io/message/InitProducerIdRequest.json

@@ -25,7 +25,9 @@
   // Version 3 adds ProducerId and ProducerEpoch, allowing producers to try to resume after an INVALID_PRODUCER_EPOCH error
   //
   // Version 4 adds the support for new error code PRODUCER_FENCED.
-  "validVersions": "0-4",
+  //
+  // Verison 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-5",
   "flexibleVersions": "2+",
   "fields": [
     { "name": "TransactionalId", "type": "string", "versions": "0+", "nullableVersions": "0+", "entityType": "transactionalId",

tansu-kafka-sans-io/message/InitProducerIdResponse.json

@@ -24,7 +24,9 @@
   // Version 3 is the same as version 2.
   //
   // Version 4 adds the support for new error code PRODUCER_FENCED.
-  "validVersions": "0-4",
+  //
+  // Version 5 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-5",
   "flexibleVersions": "2+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,

tansu-kafka-sans-io/message/ListClientMetricsResourcesResponse.json

@@ -22,9 +22,12 @@
   "fields": [
       { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
         "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
-      { "name": "ErrorCode", "type": "int16", "versions": "0+" },
-      { "name": "ClientMetricsResources", "type": "[]ClientMetricsResource", "versions": "0+", "fields": [
-        { "name": "Name", "type": "string", "versions": "0+" }
+      { "name": "ErrorCode", "type": "int16", "versions": "0+",
+        "about": "The error code, or 0 if there was no error." },
+      { "name": "ClientMetricsResources", "type": "[]ClientMetricsResource", "versions": "0+",
+        "about": "Each client metrics resource in the response.", "fields": [
+        { "name": "Name", "type": "string", "versions": "0+",
+          "about": "The resource name." }
     ]}
   ]
 }

tansu-kafka-sans-io/message/ListTransactionsRequest.json

@@ -18,14 +18,18 @@
   "type": "request",
   "listeners": ["zkBroker", "broker"],
   "name": "ListTransactionsRequest",
-  "validVersions": "0",
+  // Version 1: adds DurationFilter to list transactions older than specified duration
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "StateFilters", "type": "[]string", "versions": "0+",
       "about": "The transaction states to filter by: if empty, all transactions are returned; if non-empty, then only transactions matching one of the filtered states will be returned"
     },
     { "name": "ProducerIdFilters", "type": "[]int64", "versions": "0+", "entityType": "producerId",
       "about": "The producerIds to filter by: if empty, all transactions will be returned; if non-empty, only transactions which match one of the filtered producerIds will be returned"
+    },
+    { "name": "DurationFilter", "type": "int64", "versions": "1+", "default": -1,
+      "about": "Duration (in millis) to filter by: if < 0, all transactions will be returned; otherwise, only transactions running longer than this duration will be returned"
     }
   ]
 }

tansu-kafka-sans-io/message/ListTransactionsResponse.json

@@ -17,7 +17,8 @@
   "apiKey": 66,
   "type": "response",
   "name": "ListTransactionsResponse",
-  "validVersions": "0",
+  // Version 1 is the same as version 0 (KIP-994).
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
       { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

tansu-kafka-sans-io/message/MetadataResponse.json

@@ -48,7 +48,7 @@
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "3+", "ignorable": true,
       "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
     { "name": "Brokers", "type": "[]MetadataResponseBroker", "versions": "0+",
-      "about": "Each broker in the response.", "fields": [
+      "about": "A list of brokers present in the cluster.", "fields": [
       { "name": "NodeId", "type": "int32", "versions": "0+", "mapKey": true, "entityType": "brokerId",
         "about": "The broker ID." },
       { "name": "Host", "type": "string", "versions": "0+",
@@ -67,8 +67,9 @@
       { "name": "ErrorCode", "type": "int16", "versions": "0+",
         "about": "The topic error, or 0 if there was no error." },
       { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName", "nullableVersions": "12+",
-        "about": "The topic name." },
-      { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true, "about": "The topic id." },
+        "about": "The topic name. Null for non-existing topics queried by ID. This is never null when ErrorCode is zero. One of Name and TopicId is always populated." },
+      { "name": "TopicId", "type": "uuid", "versions": "10+", "ignorable": true,
+        "about": "The topic id. Zero for non-existing topics queried by name. This is never zero when ErrorCode is zero. One of Name and TopicId is always populated." },
       { "name": "IsInternal", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
         "about": "True if the topic is internal." },
       { "name": "Partitions", "type": "[]MetadataResponsePartition", "versions": "0+",

tansu-kafka-sans-io/message/ProduceRequest.json

@@ -35,7 +35,9 @@
   // Version 9 enables flexible versions.
   //
   // Version 10 is the same as version 9 (KIP-951).
-  "validVersions": "0-10",
+  //
+  // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-11",
   "deprecatedVersions": "0-6",
   "flexibleVersions": "9+",
   "fields": [

tansu-kafka-sans-io/message/ProduceResponse.json

@@ -34,7 +34,9 @@
   // Version 9 enables flexible versions.
   //
   // Version 10 adds 'CurrentLeader' and 'NodeEndpoints' as tagged fields (KIP-951)
-  "validVersions": "0-10",
+  //
+  // Version 11 adds support for new error code TRANSACTION_ABORTABLE (KIP-890).
+  "validVersions": "0-11",
   "flexibleVersions": "9+",
   "fields": [
     { "name": "Responses", "type": "[]TopicProduceResponse", "versions": "0+",