Repository: strimzi/strimzi-kafka-bridge
Pull request #961 (2025-01-12 17:20:24)

Refactored to use new Java version constructs
With Java 11 dropped, this PR makes some refactoring around newly constructs in modern Java.


Patch:
diff --git a/src/main/java/io/strimzi/kafka/bridge/EmbeddedFormat.java b/src/main/java/io/strimzi/kafka/bridge/EmbeddedFormat.java
index 16de02d..a8b53c9 100644
--- a/src/main/java/io/strimzi/kafka/bridge/EmbeddedFormat.java
+++ b/src/main/java/io/strimzi/kafka/bridge/EmbeddedFormat.java
@@ -26,14 +26,11 @@ public enum EmbeddedFormat {
      * @return corresponding enum
      */
     public static EmbeddedFormat from(String value) {
-        switch (value) {
-            case "json":
-                return JSON;
-            case "binary":
-                return BINARY;
-            case "text":
-                return TEXT;
-        }
-        throw new IllegalEmbeddedFormatException("Invalid format type.");
+        return switch (value) {
+            case "json" -> JSON;
+            case "binary" -> BINARY;
+            case "text" -> TEXT;
+            default -> throw new IllegalEmbeddedFormatException("Invalid format type.");
+        };
     }
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminBridgeEndpoint.java
index 8b0763e..f795efa 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminBridgeEndpoint.java
@@ -44,7 +44,7 @@ import java.util.concurrent.CompletionStage;
 public class HttpAdminBridgeEndpoint extends HttpBridgeEndpoint {
     private static final Logger LOGGER = LogManager.getLogger(HttpAdminBridgeEndpoint.class);
 
-    private final HttpBridgeContext httpBridgeContext;
+    private final HttpBridgeContext<?, ?> httpBridgeContext;
     private final KafkaBridgeAdmin kafkaBridgeAdmin;
 
     /**
@@ -53,7 +53,7 @@ public class HttpAdminBridgeEndpoint extends HttpBridgeEndpoint {
      * @param bridgeConfig the bridge configuration
      * @param context the HTTP bridge context
      */
-    public HttpAdminBridgeEndpoint(BridgeConfig bridgeConfig, HttpBridgeContext context) {
+    public HttpAdminBridgeEndpoint(BridgeConfig bridgeConfig, HttpBridgeContext<?, ?> context) {
         super(bridgeConfig);
         this.name = "kafka-bridge-admin";
         this.httpBridgeContext = context;
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
index fbebb71..68511a5 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -329,7 +329,7 @@ public class HttpBridge extends AbstractVerticle {
                 responseStatus.code(),
                 ex.getMessage()
             );
-            HttpUtils.sendResponse(routingContext, error.getCode(),
+            HttpUtils.sendResponse(routingContext, error.code(),
                     BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
         }
     }
@@ -586,20 +586,17 @@ public class HttpBridge extends AbstractVerticle {
             // in case of validation exception, building a meaningful error message
             if (routingContext.failure() != null) {
                 StringBuilder sb = new StringBuilder();
-                if (routingContext.failure().getCause() instanceof ValidationException) {
-                    ValidationException validationException = (ValidationException) routingContext.failure().getCause();
+                if (routingContext.failure().getCause() instanceof ValidationException validationException) {
                     if (validationException.inputScope() != null) {
                         sb.append("Validation error on: ").append(validationException.inputScope()).append(" - ");
                     }
                     sb.append(validationException.getMessage());
-                } else if (routingContext.failure() instanceof ParameterProcessorException) {
-                    ParameterProcessorException parameterException = (ParameterProcessorException) routingContext.failure();
+                } else if (routingContext.failure() instanceof ParameterProcessorException parameterException) {
                     if (parameterException.getParameterName() != null) {
                         sb.append("Parameter error on: ").append(parameterException.getParameterName()).append(" - ");
                     }
                     sb.append(parameterException.getMessage());
-                } else if (routingContext.failure() instanceof BodyProcessorException) {
-                    BodyProcessorException bodyProcessorException = (BodyProcessorException) routingContext.failure();
+                } else if (routingContext.failure() instanceof BodyProcessorException bodyProcessorException) {
                     sb.append(bodyProcessorException.getMessage());
                 }
                 message = sb.toString();
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 b54cb0e..dc51e1f 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -197,7 +197,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
                 HttpUtils.sendResponse(routingContext, HttpResponseStatus.NO_CONTENT.code(), null, null);
             } else {
                 HttpBridgeError error = handleError(ex);
-                HttpUtils.sendResponse(routingContext, error.getCode(),
+                HttpUtils.sendResponse(routingContext, error.code(),
                         BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
             }
         });
@@ -223,7 +223,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
                 HttpUtils.sendResponse(routingContext, HttpResponseStatus.NO_CONTENT.code(), null, null);
             } else {
                 HttpBridgeError error = handleError(ex);
-                HttpUtils.sendResponse(routingContext, error.getCode(),
+                HttpUtils.sendResponse(routingContext, error.code(),
                         BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
             }
         });
@@ -388,7 +388,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
             StreamSupport.stream(partitionsList.spliterator(), false)
                     .map(JsonNode.class::cast)
                     .map(json -> new SinkTopicSubscription(JsonUtils.getString(json, "topic"), JsonUtils.getInt(json, "partition")))
-                    .collect(Collectors.toList())
+                    .toList()
         );
 
         // fulfilling the request in a separate thread to free the Vert.x event loop still in place
@@ -441,7 +441,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
                     StreamSupport.stream(topicsList.spliterator(), false)
                             .map(TextNode.class::cast)
                             .map(topic -> new SinkTopicSubscription(topic.asText()))
-                            .collect(Collectors.toList())
+                            .toList()
                 );
                 this.kafkaBridgeConsumer.subscribe(topicSubscriptions);
             } else if (bodyAsJson.has("topic_pattern")) {
@@ -554,7 +554,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
             LOGGER.debug("[{}] Request: body = {}", routingContext.get("request-id"), bodyAsJson);
         } catch (JsonDecodeException ex) {
             HttpBridgeError error = handleError(ex);
-            HttpUtils.sendResponse(routingContext, error.getCode(),
+            HttpUtils.sendResponse(routingContext, error.code(),
                     BridgeContentType.KAFKA_JSON, JsonUtils.jsonToBytes(error.toJson()));
             return;
         }
@@ -609,39 +609,30 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
 
     @SuppressWarnings("unchecked")
     private MessageConverter<K, V, byte[], byte[]> buildMessageConverter() {
-        switch (this.format) {
-            case JSON:
-                return (MessageConverter<K, V, byte[], byte[]>) new HttpJsonMessageConverter();
-            case BINARY:
-                return (MessageConverter<K, V, byte[], byte[]>) new HttpBinaryMessageConverter();
-            case TEXT:
-                return (MessageConverter<K, V, byte[], byte[]>) new HttpTextMessageConverter();
-        }
-        return null;
+        return switch (this.format) {
+            case JSON -> (MessageConverter<K, V, byte[], byte[]>) new HttpJsonMessageConverter();
+            case BINARY -> (MessageConverter<K, V, byte[], byte[]>) new HttpBinaryMessageConverter();
+            case TEXT -> (MessageConverter<K, V, byte[], byte[]>) new HttpTextMessageConverter();
+            default -> null;
+        };
     }
 
     private String getContentType() {
-        switch (this.format) {
-            case JSON:
-                return BridgeContentType.KAFKA_JSON_JSON;
-            case BINARY:
-                return BridgeContentType.KAFKA_JSON_BINARY;
-            case TEXT:
-                return BridgeContentType.KAFKA_JSON_TEXT;
-        }
-        throw new IllegalArgumentException();
+        return switch (this.format) {
+            case JSON -> BridgeContentType.KAFKA_JSON_JSON;
+            case BINARY -> BridgeContentType.KAFKA_JSON_BINARY;
+            case TEXT -> BridgeContentType.KAFKA_JSON_TEXT;
+            default -> throw new IllegalArgumentException();
+        };
     }
 
     private boolean checkAcceptedBody(String accept) {
-        switch (accept) {
-            case BridgeContentType.KAFKA_JSON_JSON:
-                return format == EmbeddedFormat.JSON;
-            case BridgeContentType.KAFKA_JSON_BINARY:
-                return format == EmbeddedFormat.BINARY;
-            case BridgeContentType.KAFKA_JSON_TEXT:
-                return format == EmbeddedFormat.TEXT;
-        }
-        return false;
+        return switch (accept) {
+            case BridgeContentType.KAFKA_JSON_JSON -> format == EmbeddedFormat.JSON;
+            case BridgeContentType.KAFKA_JSON_BINARY -> format == EmbeddedFormat.BINARY;
+            case BridgeContentType.KAFKA_JSON_TEXT -> format == EmbeddedFormat.TEXT;
+            default -> false;
+        };
     }
 
     /**
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
index 76a8157..57d0897 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -191,13 +191,11 @@ public class HttpSourceBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
 
         for (HttpBridgeResult<?> result : results) {
             ObjectNode offset = null;
-            if (result.getResult() instanceof RecordMetadata) {
-                RecordMetadata metadata = (RecordMetadata) result.getResult();
+            if (result.result() instanceof RecordMetadata metadata) {
                 offset = JsonUtils.createObjectNode()
                         .put("partition", metadata.partition())
                         .put("offset", metadata.offset());
-            } else if (result.getResult() instanceof HttpBridgeError) {
-                HttpBridgeError error = (HttpBridgeError) result.getResult();
+            } else if (result.result() instanceof HttpBridgeError error) {
                 offset = error.toJson();
             }
             offsets.add(offset);
@@ -218,14 +216,14 @@ public class HttpSourceBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
 
     @SuppressWarnings("unchecked")
     private MessageConverter<K, V, byte[], byte[]> buildMessageConverter(String contentType) {
-        switch (contentType) {
-            case BridgeContentType.KAFKA_JSON_JSON:
-                return (MessageConverter<K, V, byte[], byte[]>) new HttpJsonMessageConverter();
-            case BridgeContentType.KAFKA_JSON_BINARY:
-                return (MessageConverter<K, V, byte[], byte[]>) new HttpBinaryMessageConverter();
-            case BridgeContentType.KAFKA_JSON_TEXT:
-                return (MessageConverter<K, V, byte[], byte[]>) new HttpTextMessageConverter();
-        }
-        return null;
+        return switch (contentType) {
+            case BridgeContentType.KAFKA_JSON_JSON ->
+                    (MessageConverter<K, V, byte[], byte[]>) new HttpJsonMessageConverter();
+            case BridgeContentType.KAFKA_JSON_BINARY ->
+                    (MessageConverter<K, V, byte[], byte[]>) new HttpBinaryMessageConverter();
+            case BridgeContentType.KAFKA_JSON_TEXT ->
+                    (MessageConverter<K, V, byte[], byte[]>) new HttpTextMessageConverter();
+            default -> null;
+        };
     }
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java
index 409a0bb..87aed91 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java
@@ -11,36 +11,11 @@ import io.vertx.core.json.JsonObject;
 
 /**
  * Represents an error related to HTTP bridging
+ *
+ * @param code  code classifying the error itself
+ * @param message   message providing more information about the error
  */
-public class HttpBridgeError {
-
-    private final int code;
-    private final String message;
-
-    /**
-     * Constructor
-     *
-     * @param code code classifying the error itself
-     * @param message message providing more information about the error
-     */
-    public HttpBridgeError(int code, String message) {
-        this.code = code;
-        this.message = message;
-    }
-
-    /**
-     * @return code classifying the error itself
-     */
-    public int getCode() {
-        return code;
-    }
-
-    /**
-     * @return message providing more information about the error
-     */
-    public String getMessage() {
-        return message;
-    }
+public record HttpBridgeError(int code, String message) {
 
     /**
      * @return a JSON representation of the error with code and message
@@ -54,6 +29,7 @@ public class HttpBridgeError {
 
     /**
      * Create an error instance from a JSON representation
+     *
      * @param json JSON representation of the error
      * @return error instance
      */
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java
index befb2a3..c0d3129 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java
@@ -9,24 +9,6 @@ package io.strimzi.kafka.bridge.http.model;
  * This class represents a result of an HTTP bridging operation
  *
  * @param <T> the class bringing the actual result as {@link HttpBridgeError} or {@link org.apache.kafka.clients.producer.RecordMetadata}
+ * @param result    actual result
  */
-public class HttpBridgeResult<T> {
-
-    final T result;
-
-    /**
-     * Constructor
-     *
-     * @param result actual result
-     */
-    public HttpBridgeResult(T result) {
-        this.result = result;
-    }
-
-    /**
-     * @return the actual result
-     */
-    public T getResult() {
-        return result;
-    }
-}
+public record HttpBridgeResult<T>(T result) { }


Test patch:
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 f9d5776..a38974e 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerIT.java
@@ -91,8 +91,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getMessage(), is("Invalid format type."));
+                        assertThat(error.code(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.message(), is("Invalid format type."));
                     });
                     create.complete(true);
                 });
@@ -153,8 +153,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getMessage(), is("Validation error on: /enable.auto.commit - input don't match type BOOLEAN"));
+                        assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                        assertThat(error.message(), is("Validation error on: /enable.auto.commit - input don't match type BOOLEAN"));
                     });
                     createBooleanAsString.complete(true);
                 });
@@ -174,8 +174,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getMessage(), is("Validation error on: /enable.auto.commit - input don't match type BOOLEAN"));
+                        assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                        assertThat(error.message(), is("Validation error on: /enable.auto.commit - input don't match type BOOLEAN"));
                     });
                     createGenericString.complete(true);
                 });
@@ -218,8 +218,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getMessage(), is("Validation error on: /" + param + " - input don't match type INTEGER"));
+                        assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                        assertThat(error.message(), is("Validation error on: /" + param + " - input don't match type INTEGER"));
                     });
                     createIntegerAsString.complete(true);
                 });
@@ -239,8 +239,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getMessage(), is("Validation error on: /" + param + " - input don't match type INTEGER"));
+                        assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                        assertThat(error.message(), is("Validation error on: /" + param + " - input don't match type INTEGER"));
                     });
                     createGenericString.complete(true);
                 });
@@ -430,8 +430,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
-                        assertThat(error.getMessage(), is("mqtt is not a valid schema/proto."));
+                        assertThat(error.code(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
+                        assertThat(error.message(), is("mqtt is not a valid schema/proto."));
                     });
                     context.completeNow();
                 });
@@ -1196,8 +1196,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.CONFLICT.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.CONFLICT.code()));
-                        assertThat(error.getMessage(), is("A consumer instance with the specified name already exists in the Kafka Bridge."));
+                        assertThat(error.code(), is(HttpResponseStatus.CONFLICT.code()));
+                        assertThat(error.message(), is("A consumer instance with the specified name already exists in the Kafka Bridge."));
                         context.completeNow();
                     });
                     create2Again.complete(true);
@@ -1244,8 +1244,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                     HttpResponse<JsonObject> response = ar.result();
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                     assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                    assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                    assertThat(error.getMessage(), is("The specified consumer instance was not found."));
+                    assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                    assertThat(error.message(), is("The specified consumer instance was not found."));
                 });
                 context.completeNow();
             });
@@ -1272,8 +1272,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("The specified consumer instance was not found."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("The specified consumer instance was not found."));
                     });
                     context.completeNow();
                 });
@@ -1309,8 +1309,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getMessage(), is("Response exceeds the maximum number of bytes the consumer can receive"));
+                        assertThat(error.code(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.message(), is("Response exceeds the maximum number of bytes the consumer can receive"));
                     });
                     consume.complete(true);
                 });
@@ -1388,8 +1388,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
-                        assertThat(error.getMessage(), is("Consumer is not subscribed to any topics or assigned any partitions"));
+                        assertThat(error.code(), is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
+                        assertThat(error.message(), is("Consumer is not subscribed to any topics or assigned any partitions"));
                     });
                     consume2.complete(true);
                 });
@@ -1429,8 +1429,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_ACCEPTABLE.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_ACCEPTABLE.code()));
-                        assertThat(error.getMessage(), is("Consumer format does not match the embedded format requested by the Accept header."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_ACCEPTABLE.code()));
+                        assertThat(error.message(), is("Consumer format does not match the embedded format requested by the Accept header."));
                     });
                     consume.complete(true);
                 });
@@ -1571,8 +1571,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_ACCEPTABLE.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_ACCEPTABLE.code()));
-                        assertThat(error.getMessage().startsWith("Failed to decode"), is(true));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_ACCEPTABLE.code()));
+                        assertThat(error.message().startsWith("Failed to decode"), is(true));
                     });
                     consume.complete(true);
                 });
@@ -1694,8 +1694,8 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                     HttpResponse<JsonObject> response = ar.result();
                     assertThat(response.statusCode(), is(status.code()));
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    assertThat(error.getCode(), is(status.code()));
-                    assertThat(error.getMessage(), is(message));
+                    assertThat(error.code(), is(status.code()));
+                    assertThat(error.message(), is(message));
                 });
                 consumer.complete(true);
             });
@@ -1722,9 +1722,9 @@ public class ConsumerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat("Response status code is not '422'", response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat("Response status code is not '422'", HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), is(error.getCode()));
-                        LOGGER.info("This is message -> " + error.getMessage());
-                        assertThat("Body message doesn't contain 'Invalid format type.'", error.getMessage(), equalTo("Invalid format type."));
+                        assertThat("Response status code is not '422'", HttpResponseStatus.UNPROCESSABLE_ENTITY.code(), is(error.code()));
+                        LOGGER.info("This is message -> " + error.message());
+                        assertThat("Body message doesn't contain 'Invalid format type.'", error.message(), equalTo("Invalid format type."));
                     });
                     create.complete(true);
                 });
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java
index 27b9d9e..4ba2c52 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionIT.java
@@ -58,8 +58,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("The specified consumer instance was not found."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("The specified consumer instance was not found."));
                     });
                     unsubscribe.complete(true);
                 });
@@ -97,8 +97,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.CONFLICT.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.CONFLICT.code()));
-                        assertThat(error.getMessage(), is("Subscriptions to topics, partitions, and patterns are mutually exclusive."));
+                        assertThat(error.code(), is(HttpResponseStatus.CONFLICT.code()));
+                        assertThat(error.message(), is("Subscriptions to topics, partitions, and patterns are mutually exclusive."));
                     });
 
                     subscribeConflict.complete(true);
@@ -117,8 +117,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getMessage(), is("A list (of Topics type) or a topic_pattern must be specified."));
+                        assertThat(error.code(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.message(), is("A list (of Topics type) or a topic_pattern must be specified."));
                     });
 
                     subscribeEmpty.complete(true);
@@ -152,8 +152,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("The specified consumer instance was not found."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("The specified consumer instance was not found."));
                     });
                     subscribe.complete(true);
                 });
@@ -277,8 +277,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("The specified consumer instance was not found."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("The specified consumer instance was not found."));
                     });
                     subscribe.complete(true);
                 });
@@ -377,8 +377,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), CoreMatchers.is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
-                        assertThat(error.getCode(), CoreMatchers.is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
-                        assertThat(error.getMessage(), CoreMatchers.is("Consumer is not subscribed to any topics or assigned any partitions"));
+                        assertThat(error.code(), CoreMatchers.is(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()));
+                        assertThat(error.message(), CoreMatchers.is("Consumer is not subscribed to any topics or assigned any partitions"));
 
                     });
                     consume.complete(true);
@@ -429,8 +429,8 @@ public class ConsumerSubscriptionIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.CONFLICT.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.CONFLICT.code()));
-                        assertThat(error.getMessage(), is("Subscriptions to topics, partitions, and patterns are mutually exclusive."));
+                        assertThat(error.code(), is(HttpResponseStatus.CONFLICT.code()));
+                        assertThat(error.message(), is("Subscriptions to topics, partitions, and patterns are mutually exclusive."));
                     });
                     assignCF.complete(true);
                 });
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java
index a951f5a..d668dd2 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java
@@ -111,8 +111,8 @@ public class DisablingConsumerProducerIT {
                     HttpResponse<JsonObject> response = ar.result();
                     assertThat(response.statusCode(), CoreMatchers.is(HttpResponseStatus.SERVICE_UNAVAILABLE.code()));
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    assertThat(error.getCode(), CoreMatchers.is(HttpResponseStatus.SERVICE_UNAVAILABLE.code()));
-                    assertThat(error.getMessage(), CoreMatchers.is("Consumer is disabled in config. To enable consumer update http.consumer.enabled to true"));
+                    assertThat(error.code(), CoreMatchers.is(HttpResponseStatus.SERVICE_UNAVAILABLE.code()));
+                    assertThat(error.message(), CoreMatchers.is("Consumer is disabled in config. To enable consumer update http.consumer.enabled to true"));
                     tryCreate.complete(true);
                 });
             });
@@ -148,8 +148,8 @@ public class DisablingConsumerProducerIT {
                     HttpResponse<JsonObject> response = ar.result();
                     assertThat(response.statusCode(), CoreMatchers.is(HttpResponseStatus.SERVICE_UNAVAILABLE.code()));
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                    assertThat(error.getCode(), CoreMatchers.is(HttpResponseStatus.SERVICE_UNAVAILABLE.code()));
-                    assertThat(error.getMessage(), CoreMatchers.is("Producer is disabled in config. To enable producer update http.producer.enabled to true"));
+                    assertThat(error.code(), CoreMatchers.is(HttpResponseStatus.SERVICE_UNAVAILABLE.code()));
+                    assertThat(error.message(), CoreMatchers.is("Producer is disabled in config. To enable producer update http.producer.enabled to true"));
                     trySend.complete(true);
                 });
             });
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesIT.java b/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesIT.java
index b4b7799..b6afec1 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesIT.java
@@ -92,8 +92,8 @@ public class OtherServicesIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.GONE.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.GONE.code()));
-                        assertThat(error.getMessage(), is("OpenAPI v2 Swagger not supported"));
+                        assertThat(error.code(), is(HttpResponseStatus.GONE.code()));
+                        assertThat(error.message(), is("OpenAPI v2 Swagger not supported"));
                     });
                     context.completeNow();
                 });
@@ -193,7 +193,7 @@ public class OtherServicesIT extends HttpBridgeITAbstract {
                     HttpResponse<JsonObject> response = ar.result();
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                     assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                    assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
+                    assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
                 });
                 context.completeNow();
             });
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java
index 37fd5eb..5004bf2 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java
@@ -706,7 +706,7 @@ public class ProducerIT extends HttpBridgeITAbstract {
                     HttpResponse<JsonObject> response = ar.result();
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                     assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                    assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                    assertThat(error.code(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
                 });
                 context.completeNow();
             });
@@ -737,8 +737,8 @@ public class ProducerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getMessage(), is("Because the embedded format is 'text', the value must be a string"));
+                        assertThat(error.code(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.message(), is("Because the embedded format is 'text', the value must be a string"));
                     });
                     context.completeNow();
                 });
@@ -770,8 +770,8 @@ public class ProducerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
-                        assertThat(error.getMessage(), is("Because the embedded format is 'text', the key must be a string"));
+                        assertThat(error.code(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.message(), is("Because the embedded format is 'text', the key must be a string"));
                     });
                     context.completeNow();
                 });
@@ -853,10 +853,10 @@ public class ProducerIT extends HttpBridgeITAbstract {
                     assertThat(offsets.size(), is(1));
 
                     HttpBridgeError error = HttpBridgeError.fromJson(offsets.getJsonObject(0));
-                    assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
+                    assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
                     // the message got from the Kafka producer (starting from 2.3) is misleading
                     // this JIRA (https://issues.apache.org/jira/browse/KAFKA-8862) raises the issue
-                    assertThat(error.getMessage(), is(
+                    assertThat(error.message(), is(
                             "Topic " + topic + " not present in metadata after " +
                                     config.get(KafkaProducerConfig.KAFKA_PRODUCER_CONFIG_PREFIX + ProducerConfig.MAX_BLOCK_MS_CONFIG) + " ms."));
                 });
@@ -1036,8 +1036,8 @@ public class ProducerIT extends HttpBridgeITAbstract {
                 HttpResponse<JsonObject> response = ar.result();
                 assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
                 HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                assertThat(error.getMessage(), is(message));
+                assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                assertThat(error.message(), is(message));
                 context.completeNow();
             });
     }
@@ -1094,8 +1094,8 @@ public class ProducerIT extends HttpBridgeITAbstract {
                         assertThat(metadata.getInteger("partition"), is(partition));
                         assertThat(metadata.getLong("offset"), is(0L));
                         HttpBridgeError error = HttpBridgeError.fromJson(offsets.getJsonObject(1));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("Topic " + topic + " not present in metadata after 10000 ms."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("Topic " + topic + " not present in metadata after 10000 ms."));
                     });
                     context.completeNow();
                 });
@@ -1272,8 +1272,8 @@ public class ProducerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getMessage(), containsString("Value wrong should be true or false"));
+                        assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                        assertThat(error.message(), containsString("Value wrong should be true or false"));
                     });
                     context.completeNow();
                 });
@@ -1305,8 +1305,8 @@ public class ProducerIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.BAD_REQUEST.code()));
-                        assertThat(error.getMessage(), containsString("Cannot find body processor for content type"));
+                        assertThat(error.code(), is(HttpResponseStatus.BAD_REQUEST.code()));
+                        assertThat(error.message(), containsString("Cannot find body processor for content type"));
                     });
                     context.completeNow();
                 });
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/SeekIT.java b/src/test/java/io/strimzi/kafka/bridge/http/SeekIT.java
index a8f2f88..70a5894 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/SeekIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/SeekIT.java
@@ -52,8 +52,8 @@ public class SeekIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("The specified consumer instance was not found."));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("The specified consumer instance was not found."));
                         context.completeNow();
                     });
                 });
@@ -84,8 +84,8 @@ public class SeekIT extends HttpBridgeITAbstract {
                     HttpResponse<JsonObject> response = ar.result();
                     HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                     assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                    assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                    assertThat(error.getMessage(), is("No current assignment for partition " + topic + "-" + notExistingPartition));
+                    assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                    assertThat(error.message(), is("No current assignment for partition " + topic + "-" + notExistingPartition));
                     context.completeNow();
                 });
             });
@@ -119,8 +119,8 @@ public class SeekIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("No current assignment for partition " + notExistingTopic + "-" + 0));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("No current assignment for partition " + notExistingTopic + "-" + 0));
                         context.completeNow();
                     });
                     consumerInstanceDontHaveTopic.complete(true);
@@ -446,8 +446,8 @@ public class SeekIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("No current assignment for partition " + notSubscribedTopic + "-0"));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("No current assignment for partition " + notSubscribedTopic + "-0"));
                     });
                     seek.complete(true);
                 });
@@ -514,8 +514,8 @@ public class SeekIT extends HttpBridgeITAbstract {
                         HttpResponse<JsonObject> response = ar.result();
                         HttpBridgeError error = HttpBridgeError.fromJson(response.body());
                         assertThat(response.statusCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getCode(), is(HttpResponseStatus.NOT_FOUND.code()));
-                        assertThat(error.getMessage(), is("No current assignment for partition " + notSubscribedTopic + "-0"));
+                        assertThat(error.code(), is(HttpResponseStatus.NOT_FOUND.code()));
+                        assertThat(error.message(), is("No current assignment for partition " + notSubscribedTopic + "-0"));
                     });
                     seek.complete(true);
                 });