Make sure that ErrorMessage is always interpreted as json in Response.

Integration to show that without explicit MediaType, Options requests can make server fails with 500

core/src/main/java/io/confluent/rest/exceptions/ConstraintViolationExceptionMapper.java

@@ -17,6 +17,7 @@
 package io.confluent.rest.exceptions;
 
 import javax.validation.ConstraintViolationException;
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.ExceptionMapper;
 import javax.ws.rs.ext.Provider;
@@ -67,6 +68,7 @@ public Response toResponse(ConstraintViolationException exception) {
 
     return Response.status(UNPROCESSABLE_ENTITY_CODE)
         .entity(message)
+        .type(MediaType.APPLICATION_JSON_TYPE)
         .build();
   }
 }

core/src/main/java/io/confluent/rest/exceptions/DebuggableExceptionMapper.java

@@ -23,6 +23,7 @@
 
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.ExceptionMapper;
 import javax.ws.rs.ext.Provider;
@@ -118,7 +119,8 @@ public Response.ResponseBuilder createResponse(Throwable exc, int errorCode,
     final ErrorMessage message = new ErrorMessage(errorCode, readableMessage);
 
     return Response.status(status)
-        .entity(message);
+        .entity(message)
+        .type(MediaType.APPLICATION_JSON_TYPE);
   }
 
 }

core/src/main/java/io/confluent/rest/exceptions/JsonMappingExceptionMapper.java

@@ -22,6 +22,7 @@
 import io.confluent.rest.entities.ErrorMessage;
 
 import javax.annotation.Priority;
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.ExceptionMapper;
 import javax.ws.rs.ext.Provider;
@@ -47,6 +48,7 @@ public Response toResponse(JsonMappingException exception) {
 
     return Response.status(BAD_REQUEST_CODE)
         .entity(message)
+        .type(MediaType.APPLICATION_JSON_TYPE)
         .build();
   }
 

core/src/main/java/io/confluent/rest/exceptions/JsonParseExceptionMapper.java

@@ -20,6 +20,7 @@
 import io.confluent.rest.entities.ErrorMessage;
 
 import javax.annotation.Priority;
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.ExceptionMapper;
 import javax.ws.rs.ext.Provider;
@@ -40,6 +41,7 @@ public Response toResponse(JsonParseException exception) {
 
     return Response.status(BAD_REQUEST_CODE)
         .entity(message)
+        .type(MediaType.APPLICATION_JSON_TYPE)
         .build();
   }
 }

core/src/main/java/io/confluent/rest/exceptions/KafkaExceptionMapper.java

@@ -18,6 +18,7 @@
 
 import io.confluent.rest.RestConfig;
 import io.confluent.rest.entities.ErrorMessage;
+import javax.ws.rs.core.MediaType;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.errors.AuthenticationException;
@@ -153,14 +154,18 @@ private Response getResponse(final Throwable exception, final Status status,
                                final int errorCode) {
     ErrorMessage errorMessage = new ErrorMessage(errorCode, exception.getMessage());
     return Response.status(status)
-        .entity(errorMessage).build();
+        .entity(errorMessage)
+        .type(MediaType.APPLICATION_JSON_TYPE)
+        .build();
   }
 
   private Response getResponse(final Throwable cause) {
     ResponsePair responsePair = HANDLED.get(cause.getClass());
     ErrorMessage errorMessage = new ErrorMessage(responsePair.errorCode, cause.getMessage());
     return Response.status(responsePair.status)
-        .entity(errorMessage).build();
+        .entity(errorMessage)
+        .type(MediaType.APPLICATION_JSON_TYPE)
+        .build();
   }
 
   private static class ResponsePair {

core/src/test/java/io/confluent/rest/ErrorMessageIntegrationTest.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2014 - 2023 Confluent Inc.
+ *
+ * Licensed 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.
+ */
+
+package io.confluent.rest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import com.fasterxml.jackson.core.JsonLocation;
+import com.fasterxml.jackson.core.JsonParseException;
+import io.confluent.rest.entities.ErrorMessage;
+import io.confluent.rest.exceptions.JsonParseExceptionMapper;
+import java.util.Properties;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.container.ContainerRequestContext;
+import javax.ws.rs.container.ContainerRequestFilter;
+import javax.ws.rs.core.Configurable;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import org.eclipse.jetty.server.Server;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+@Tag("IntegrationTest")
+public class ErrorMessageIntegrationTest {
+
+  private Server server;
+  private Client client;
+
+  @BeforeEach
+  public void setUp(TestInfo testInfo) throws Exception {
+    Properties props = new Properties();
+    TestRestConfig config = new TestRestConfig(props);
+    TestApplication app = new TestApplication(config,
+        testInfo.getDisplayName().contains("WithJsonMediaType"));
+    app.createServer();
+    server = app.createServer();
+    server.start();
+    client = ClientBuilder.newClient(app.resourceConfig.getConfiguration());
+  }
+
+  @AfterEach
+  public void tearDown() throws Exception {
+    server.stop();
+    server.join();
+    client.close();
+  }
+
+  @Test
+  @DisplayName("testOPTIONS_WithoutMediaType_return500")
+  public void testOPTIONS_WithoutMediaType_return500() {
+    Response response = client.target(server.getURI())
+        .path("unauthorized/path")
+        .request("*/*")
+        .options();
+
+    assertEquals(Status.INTERNAL_SERVER_ERROR.getStatusCode(), response.getStatus());
+  }
+
+  @Test
+  public void testGET_WithoutMediaType_return401() {
+    Response response = client.target(server.getURI())
+        .path("unauthorized/path")
+        .request("*/*")
+        .get();
+
+    assertEquals(Status.UNAUTHORIZED.getStatusCode(), response.getStatus());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
+  }
+
+  @Test
+  @DisplayName("testOPTIONS_WithJsonMediaType_return401")
+  public void testOPTIONS_WithJsonMediaType_return401() {
+    Response response = client.target(server.getURI())
+        .path("unauthorized/path")
+        .request("*/*")
+        .options();
+
+    assertEquals(Status.UNAUTHORIZED.getStatusCode(), response.getStatus());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
+  }
+
+  @Test
+  public void testJsonParseExceptionMapper_return400() {
+    Response response = client.target(server.getURI())
+        .path("pass/json")
+        .request("*/*")
+        .get();
+
+    assertEquals(Status.BAD_REQUEST.getStatusCode(), response.getStatus());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
+  }
+
+  private static class UnauthorizedFilter implements ContainerRequestFilter {
+
+    private final boolean withJsonMediaType;
+
+    UnauthorizedFilter(boolean withJsonMediaType) {
+      this.withJsonMediaType = withJsonMediaType;
+    }
+
+    @Override
+    public void filter(ContainerRequestContext context) {
+      if (context.getUriInfo().getPath().startsWith("unauthorized/")) {
+        context.abortWith(
+            errorResponseWithEntity(Status.UNAUTHORIZED.getStatusCode(), "Unauthorized")
+        );
+      }
+    }
+
+    private Response errorResponseWithEntity(int statusCode, String message) {
+      ErrorMessage errorMessage = new ErrorMessage(statusCode, message);
+      Response.ResponseBuilder builder = Response.status(statusCode)
+          .entity(errorMessage);
+      if (withJsonMediaType) {
+        builder.type(MediaType.APPLICATION_JSON_TYPE);
+      }
+      return builder.build();
+    }
+  }
+
+  private static class TestApplication extends Application<TestRestConfig> {
+
+    Configurable<?> resourceConfig;
+    private boolean withJsonMediaType = false;
+
+    TestApplication(TestRestConfig restConfig, boolean withJsonMediaType) {
+      super(restConfig);
+      this.withJsonMediaType = withJsonMediaType;
+    }
+
+    @Override
+    public void setupResources(Configurable<?> config, TestRestConfig appConfig) {
+      resourceConfig = config;
+      config.register(TestResource.class);
+      config.register(JsonResource.class);
+      config.register(new UnauthorizedFilter(withJsonMediaType));
+      config.register(JsonParseExceptionMapper.class);
+    }
+  }
+
+  @Produces(MediaType.APPLICATION_JSON)
+  @Path("/unauthorized")
+  public static class TestResource {
+
+    @GET
+    @Path("/path")
+    public String path() {
+      return "Ok";
+    }
+  }
+
+  @Produces(MediaType.APPLICATION_JSON)
+  @Path("/pass")
+  public static class JsonResource {
+
+    @GET
+    @Path("/json")
+    public String path() throws JsonParseException {
+      throw new JsonParseException("Json parse error", JsonLocation.NA);
+    }
+  }
+}

core/src/test/java/io/confluent/rest/JsonMappingExceptionMapperTest.java

@@ -22,6 +22,7 @@
 import io.confluent.rest.entities.ErrorMessage;
 import io.confluent.rest.exceptions.JsonMappingExceptionMapper;
 
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -51,6 +52,7 @@ public void testJsonMappingException() {
       assertEquals(400, response.getStatus());
       ErrorMessage out = (ErrorMessage) response.getEntity();
       assertEquals(400, out.getErrorCode());
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
     } catch (Exception e) {
       fail("A JsonMappingException is expected.");
     }
@@ -67,6 +69,7 @@ public void testJsonMappingExceptionRemoveDetailsFromMessage() {
     ErrorMessage out = (ErrorMessage) response.getEntity();
     assertEquals(400, out.getErrorCode());
     assertEquals("Json mapping error", out.getMessage());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
   }
 
   @Test
@@ -81,6 +84,7 @@ public void testJsonMappingExceptionRemoveDetailsCannotConstruct() {
     ErrorMessage out = (ErrorMessage) response.getEntity();
     assertEquals(400, out.getErrorCode());
     assertEquals("Cannot construct instance of `CreateAclRequest`, problem: Null resourceType", out.getMessage());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
   }
 
   @Test
@@ -99,6 +103,7 @@ public void testJsonMappingExceptionRemoveDetailsCannotDeserializeEnum() {
         "values accepted for Enum class: [TRANSACTIONAL_ID, DELEGATION_TOKEN, UNKNOWN, ANY, " +
         "GROUP, CLUSTER, TOPIC]",
         out.getMessage());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
   }
 
   @Test

core/src/test/java/io/confluent/rest/JsonParseExceptionMapperTest.java

@@ -21,6 +21,7 @@
 import io.confluent.rest.entities.ErrorMessage;
 import io.confluent.rest.exceptions.JsonParseExceptionMapper;
 
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -46,5 +47,6 @@ public void testJsonParseExceptionRemoveDetailsFromMessage() {
       ErrorMessage out = (ErrorMessage)response.getEntity();
       assertEquals(400, out.getErrorCode());
       assertEquals("Json parse error", out.getMessage());
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
   }
 }

core/src/test/java/io/confluent/rest/KafkaExceptionMapperTest.java

@@ -18,6 +18,7 @@
 
 import io.confluent.rest.entities.ErrorMessage;
 import io.confluent.rest.exceptions.KafkaExceptionMapper;
+import javax.ws.rs.core.MediaType;
 import org.apache.kafka.clients.consumer.CommitFailedException;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.AuthorizationException;
@@ -172,5 +173,6 @@ private void verifyMapperResponse(Throwable throwable, Status status, int errorC
     assertEquals(status.getStatusCode(), response.getStatus());
     ErrorMessage errorMessage = (ErrorMessage) response.getEntity();
     assertEquals(errorCode, errorMessage.getErrorCode());
+    assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getMediaType());
   }
 }

core/src/test/java/io/confluent/rest/metrics/MetricsResourceMethodApplicationListenerIntegrationTest.java

@@ -709,7 +709,9 @@ public MyExceptionMapper(final RestConfig restConfig) {
     public Response toResponse(Throwable exception) {
       if (exception instanceof StatusCodeException) {
         return Response.status(Status.TOO_MANY_REQUESTS)
-            .entity(new ErrorMessage(429, exception.getMessage())).build();
+            .entity(new ErrorMessage(429, exception.getMessage()))
+            .type(MediaType.APPLICATION_JSON_TYPE)
+            .build();
       } else {
         return super.toResponse(exception); // Need this to ensure return 500 for 5XX test
       }