Added the catch of wrong content-type returning meaningful message

src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java

@@ -33,6 +33,7 @@
 
 import io.vertx.ext.web.handler.CorsHandler;
 import io.vertx.ext.web.openapi.RouterBuilder;
+import io.vertx.ext.web.validation.BodyProcessorException;
 import io.vertx.ext.web.validation.ParameterProcessorException;
 import io.vertx.json.schema.ValidationException;
 import org.apache.kafka.clients.CommonClientConfigs;
@@ -574,14 +575,21 @@ private void errorHandler(RoutingContext routingContext) {
             }
 
             if (routingContext.failure() != null && routingContext.failure() instanceof ParameterProcessorException) {
-                ParameterProcessorException validationException = (ParameterProcessorException) routingContext.failure();
+                ParameterProcessorException parameterException = (ParameterProcessorException) routingContext.failure();
                 StringBuilder sb = new StringBuilder();
-                if (validationException.getParameterName() != null) {
-                    sb.append("Validation error on: ").append(validationException.getParameterName()).append(" - ");
+                if (parameterException.getParameterName() != null) {
+                    sb.append("Parameter error on: ").append(parameterException.getParameterName()).append(" - ");
                 }
-                sb.append(validationException.getMessage());
+                sb.append(parameterException.getMessage());
                 message = sb.toString();
             }
+
+            if (routingContext.failure() != null && routingContext.failure() instanceof BodyProcessorException) {
+                BodyProcessorException bodyProcessorException = (BodyProcessorException) routingContext.failure();
+                if (bodyProcessorException.getMessage() != null) {
+                    message = bodyProcessorException.getMessage();
+                }
+            }
         } else if (routingContext.statusCode() == HttpResponseStatus.NOT_FOUND.code()) {
             message = HttpResponseStatus.NOT_FOUND.reasonPhrase();
         }

src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java

@@ -939,4 +939,37 @@ void sendAsyncMessages(VertxTestContext context) throws InterruptedException, Ex
 
         assertThat(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS), is(true));
     }
+
+    @Test
+    void sendSimpleMessageWithWrongContentType(VertxTestContext context) throws InterruptedException, ExecutionException {
+        KafkaFuture<Void> future = adminClientFacade.createTopic(topic);
+
+        String value = "message-value";
+
+        JsonArray records = new JsonArray();
+        JsonObject json = new JsonObject();
+        json.put("value", value);
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
+
+        future.get();
+
+        producerService()
+                .sendRecordsRequest(topic, root, "bad-content-type")
+                .sendJsonObject(root, ar -> {
+                    context.verify(() -> {
+                        assertThat(ar.succeeded(), is(true));
+                        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"));
+                    });
+                    context.completeNow();
+                });
+
+        assertThat(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS), is(true));
+    }
 }