Repository: strimzi/strimzi-kafka-bridge
Pull request #585 (2022-01-28 19:47:38)

Added `isolation.level` to read  commited / all transactional messages
Signed-off-by: ShubhamRwt <srawat@redhat.com>

This PR fixes the issue mentioned in #577. Now isolation level can be set to `read_committed` to see the commited transaction records or it can set to `read_uncommited` to read all transaction records.


Patch:
diff --git a/.azure/build-pipeline.yaml b/.azure/build-pipeline.yaml
index 300406b..b677ca2 100644
--- a/.azure/build-pipeline.yaml
+++ b/.azure/build-pipeline.yaml
@@ -33,7 +33,7 @@ stages:
           artifactPipeline: ''
           artifactRunVersion: ''
           artifactRunId: ''
-          architectures: ['amd64', 'arm64']
+          architectures: ['amd64', 'arm64', 's390x']
   - stage: container_publish
     displayName: Publish Container
     dependsOn:
@@ -49,7 +49,7 @@ stages:
           artifactPipeline: ''
           artifactRunVersion: ''
           artifactRunId: ''
-          architectures: ['amd64', 'arm64']
+          architectures: ['amd64', 'arm64', 's390x']
   - stage: docs_publish
     displayName: Publish Docs
     dependsOn:
diff --git a/.azure/cve-pipeline.yaml b/.azure/cve-pipeline.yaml
index f5202ff..f08e2a9 100644
--- a/.azure/cve-pipeline.yaml
+++ b/.azure/cve-pipeline.yaml
@@ -36,7 +36,7 @@ stages:
           artifactPipeline: '${{ parameters.sourcePipelineId }}'
           artifactRunVersion: 'specific'
           artifactRunId: '${{ parameters.sourceBuildId }}'
-          architectures: ['amd64', 'arm64']
+          architectures: ['amd64', 'arm64', 's390x']
   - stage: containers_publish_with_suffix
     displayName: Publish Containers for ${{ parameters.releaseVersion }}-${{ parameters.releaseSuffix }}
     dependsOn: 
@@ -51,7 +51,7 @@ stages:
           artifactPipeline: ''
           artifactRunVersion: ''
           artifactRunId: ''
-          architectures: ['amd64', 'arm64']
+          architectures: ['amd64', 'arm64', 's390x']
   - stage: manual_validation
     displayName: Validate container before pushing container as ${{ parameters.releaseVersion }}
     dependsOn:
diff --git a/.azure/release-pipeline.yaml b/.azure/release-pipeline.yaml
index 38b1f9f..97660f5 100644
--- a/.azure/release-pipeline.yaml
+++ b/.azure/release-pipeline.yaml
@@ -38,7 +38,7 @@ stages:
           artifactPipeline: '${{ parameters.sourcePipelineId }}'
           artifactRunVersion: 'specific'
           artifactRunId: '${{ parameters.sourceBuildId }}'
-          architectures: ['amd64', 'arm64']
+          architectures: ['amd64', 'arm64', 's390x']
   - stage: containers_publish
     displayName: Publish Containers for ${{ parameters.releaseVersion }}
     dependsOn:
@@ -53,4 +53,4 @@ stages:
           artifactPipeline: '${{ parameters.sourcePipelineId }}'
           artifactRunVersion: 'specific'
           artifactRunId: '${{ parameters.sourceBuildId }}'
-          architectures: ['amd64', 'arm64']
+          architectures: ['amd64', 'arm64', 's390x']
diff --git a/.travis.yml b/.travis.yml
index 10ac3b6..9954b74 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -17,7 +17,7 @@ jobs:
     os: linux
     arch: s390x
     jdk: openjdk11
-    script: mvn install -DskipTests -Dmaven.javadoc.skip=true
+    script: mvn install -Dmaven.javadoc.skip=true
     addons:
       apt:
         packages:
diff --git a/documentation/book/api/definitions.adoc b/documentation/book/api/definitions.adoc
index b60c727..70e19eb 100644
--- a/documentation/book/api/definitions.adoc
+++ b/documentation/book/api/definitions.adoc
@@ -38,6 +38,8 @@ __optional__|If set to `true`, message offsets are committed automatically for t
 __optional__|Sets the minimum ammount of data, in bytes, for the consumer to receive. The broker waits until the data to send exceeds this amount.|integer
 |**format** +
 __optional__|The allowable message format for the consumer, which can be `binary` (default) or `json`. The messages are converted into a JSON format.|string
+|**isolation.level** +
+__optional__|If set to `read_uncommitted`, all transaction records are retrieved. If set to `read_committed` , the records from committed transactions are retrieved.|string
 |**name** +
 __optional__|The unique name for the consumer instance. The name is unique within the scope of the consumer group. The name is used in URLs.|string
 |===
diff --git a/documentation/book/api/paths.adoc b/documentation/book/api/paths.adoc
index 05a0a19..d662cc0 100644
--- a/documentation/book/api/paths.adoc
+++ b/documentation/book/api/paths.adoc
@@ -90,7 +90,8 @@ __required__|Name and configuration of the consumer. The name is unique within t
   "auto.offset.reset" : "earliest",
   "enable.auto.commit" : false,
   "fetch.min.bytes" : 512,
-  "consumer.request.timeout.ms" : 30000
+  "consumer.request.timeout.ms" : 30000,
+  "isolation.level" : "read_committed"
 }
 ----
 
diff --git a/pom.xml b/pom.xml
index c7bd6a0..fbd4a58 100644
--- a/pom.xml
+++ b/pom.xml
@@ -109,7 +109,7 @@
 		<micrometer.version>1.3.9</micrometer.version>
 		<jmx-prometheus-collector.version>0.12.0</jmx-prometheus-collector.version>
 		<commons-cli.version>1.4</commons-cli.version>
-		<test-container.version>0.25.0</test-container.version>
+		<test-container.version>0.100.0</test-container.version>
 		<jakarta.version>2.3.2</jakarta.version>
 		<!-- property to skip surefire tests during failsafe execution -->
 		<!--suppress UnresolvedMavenProperty -->
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
index 8d832b3..ab5bc9f 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -131,6 +131,9 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         addConfigParameter(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,
             requestTimeoutMs != null ? String.valueOf(requestTimeoutMs) : null, config);
         addConfigParameter(ConsumerConfig.CLIENT_ID_CONFIG, this.name, config);
+        Object isolationLevel = bodyAsJson.getValue(ConsumerConfig.ISOLATION_LEVEL_CONFIG);
+        addConfigParameter(ConsumerConfig.ISOLATION_LEVEL_CONFIG,
+                isolationLevel != null ? String.valueOf(isolationLevel) : null, config);
 
         // create the consumer
         this.initConsumer(false, config);
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index 3d4289c..14a4721 100644
--- a/src/main/resources/openapi.json
+++ b/src/main/resources/openapi.json
@@ -1596,6 +1596,10 @@
                     "enable.auto.commit": {
                         "description": "If set to `true`, message offsets are committed automatically for the consumer. If set to `false`, message offsets must be committed manually.",
                         "type": "boolean"
+                    },
+                    "isolation.level": {
+                        "description": "If set to `read_uncommitted`, all transaction records are retrieved, independently on the transaction outcome (if any), and `read_committed` to get the records from committed transactions.",
+                        "type": "string"
                     }
                 },
                 "additionalProperties": false,
@@ -1605,7 +1609,8 @@
                     "auto.offset.reset": "earliest",
                     "enable.auto.commit": false,
                     "fetch.min.bytes": 512,
-                    "consumer.request.timeout.ms": 30000
+                    "consumer.request.timeout.ms": 30000,
+                    "isolation.level": "read_committed"
                 }
             },
             "OffsetCommitSeek": {
diff --git a/src/main/resources/openapiv2.json b/src/main/resources/openapiv2.json
index e093215..3c1816f 100644
--- a/src/main/resources/openapiv2.json
+++ b/src/main/resources/openapiv2.json
@@ -1439,6 +1439,10 @@
         "enable.auto.commit": {
           "description": "If set to `true`, message offsets are committed automatically for the consumer. If set to `false`, message offsets must be committed manually.",
           "type": "boolean"
+        },
+        "isolation.level": {
+          "description": "If set to `read_uncommitted`, all transaction records are retrieved. If set to `read_committed` , the records from committed transactions are retrieved.",
+          "type": "string"
         }
       },
       "additionalProperties": false,
@@ -1448,7 +1452,8 @@
         "auto.offset.reset": "earliest",
         "enable.auto.commit": false,
         "fetch.min.bytes": 512,
-        "consumer.request.timeout.ms": 30000
+        "consumer.request.timeout.ms": 30000,
+        "isolation.level": "read_committed"
       }
     },
     "OffsetCommitSeek": {


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java
index a8f001b..f2fbe2b 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java
@@ -6,7 +6,7 @@ package io.strimzi.kafka.bridge.http;
 
 import io.micrometer.core.instrument.MeterRegistry;
 import io.netty.handler.codec.http.HttpResponseStatus;
-import io.strimzi.StrimziKafkaContainer;
+import io.strimzi.test.container.StrimziKafkaContainer;
 import io.strimzi.kafka.bridge.HealthChecker;
 import io.strimzi.kafka.bridge.JmxCollectorRegistry;
 import io.strimzi.kafka.bridge.MetricsReporter;
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
index a1308f7..429cfd7 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
@@ -434,6 +434,15 @@ public class ConsumerIT extends HttpBridgeITAbstract {
         assertThat(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS), is(true));
     }
 
+    @Test
+    void createConsumerWithWrongIsolationLevel(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        checkCreatingConsumer("isolation.level", "foo", HttpResponseStatus.UNPROCESSABLE_ENTITY,
+                "Invalid value foo for configuration isolation.level: String must be one of: read_committed, read_uncommitted", context);
+
+        context.completeNow();
+        assertThat(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS), is(true));
+    }
+
     @Test
     void createConsumerWithWrongAutoOffsetReset(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
         checkCreatingConsumer("auto.offset.reset", "foo", HttpResponseStatus.UNPROCESSABLE_ENTITY,
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
index 5ca94c7..1872d64 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
@@ -6,7 +6,7 @@
 package io.strimzi.kafka.bridge.http;
 
 import io.micrometer.core.instrument.MeterRegistry;
-import io.strimzi.StrimziKafkaContainer;
+import io.strimzi.test.container.StrimziKafkaContainer;
 import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.HealthChecker;
 import io.strimzi.kafka.bridge.JmxCollectorRegistry;
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java b/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java
index b0d311b..1f53da3 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java
@@ -5,7 +5,7 @@
 
 package io.strimzi.kafka.bridge.http.base;
 
-import io.strimzi.StrimziKafkaContainer;
+import io.strimzi.test.container.StrimziKafkaContainer;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.strimzi.kafka.bridge.HealthChecker;
 import io.strimzi.kafka.bridge.JmxCollectorRegistry;