CASSANDRASC-36: Support for ErrorHandler in Sidecar

This commit adds a default `ErrorHandler` to the `Route#failureHandler()`. The default implementation
for Sidecar is `ErrorHandlerImpl`.

Additionally, we allow for custom implementations of the `ErrorHandler` to be injected (via module
overrides). This allows downstream projects to provide custom implementations of the `ErrorHandler` to
fit the specific needs of the project.

Patch by Francisco Guerrero; Reviewed by Dinesh Joshi, Saranya Krishnakumar and Yifan Cai for CASSANDRASC-36

cassandra-integration-tests/src/test/java/org/apache/cassandra/sidecar/common/testing/CassandraPod.java

@@ -339,9 +339,9 @@ private void deleteService()
             coreV1Api.deleteNamespacedService(podName, namespace, null, null, null, null, null, null);
 
         }
-        catch (Exception ignored)
+        catch (Exception ex)
         {
-            logger.info("Could not delete service {}", podName);
+            logger.info(String.format("Could not delete service %s", podName), ex);
         }
     }
 
@@ -356,9 +356,9 @@ private void deletePod()
             logger.info("Deleting pod {}", podName);
             coreV1Api.deleteNamespacedPod(podName, namespace, null, null, null, null, null, null);
         }
-        catch (Exception ignored)
+        catch (Exception ex)
         {
-            logger.error("Exception when stopping pod: {}", ignored.getMessage());
+            logger.error(String.format("Unable to delete pod %s", podName), ex);
         }
     }
 }

src/main/java/org/apache/cassandra/sidecar/MainModule.java

@@ -46,6 +46,7 @@
 import io.vertx.core.net.JksOptions;
 import io.vertx.ext.dropwizard.DropwizardMetricsOptions;
 import io.vertx.ext.web.Router;
+import io.vertx.ext.web.handler.ErrorHandler;
 import io.vertx.ext.web.handler.LoggerHandler;
 import io.vertx.ext.web.handler.StaticHandler;
 import org.apache.cassandra.sidecar.cassandra40.Cassandra40Factory;
@@ -129,10 +130,12 @@ private VertxRequestHandler configureServices(Vertx vertx,
 
     @Provides
     @Singleton
-    public Router vertxRouter(Vertx vertx, LoggerHandler loggerHandler)
+    public Router vertxRouter(Vertx vertx, LoggerHandler loggerHandler, ErrorHandler errorHandler)
     {
         Router router = Router.router(vertx);
-        router.route().handler(loggerHandler);
+        router.route()
+              .failureHandler(errorHandler)
+              .handler(loggerHandler);
 
         // Static web assets for Swagger
         StaticHandler swaggerStatic = StaticHandler.create("META-INF/resources/webjars/swagger-ui");
@@ -261,4 +264,11 @@ public LoggerHandler loggerHandler()
     {
         return LoggerHandler.create();
     }
+
+    @Provides
+    @Singleton
+    public ErrorHandler errorHandler(Vertx vertx)
+    {
+        return ErrorHandler.create(vertx);
+    }
 }

src/test/java/org/apache/cassandra/sidecar/LoggerHandlerInjectionTest.java

@@ -1,8 +1,10 @@
 package org.apache.cassandra.sidecar;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
@@ -27,76 +29,100 @@
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
 import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
 import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+/**
+ * Tests for the {@link LoggerHandler} injection tests
+ */
 @DisplayName("LoggerHandler Injection Test")
 @ExtendWith(VertxExtension.class)
 public class LoggerHandlerInjectionTest
 {
     private Vertx vertx;
     private Configuration config;
     private final Logger logger = mock(Logger.class);
+    private HttpServer server;
 
     @BeforeEach
     void setUp() throws InterruptedException
     {
         FakeLoggerHandler loggerHandler = new FakeLoggerHandler(logger);
-        Injector injector = Guice.createInjector(Modules.override(Modules.override(new MainModule()).with(new TestModule()))
-                                                        .with(binder -> binder.bind(LoggerHandler.class).toInstance(loggerHandler)));
+        Injector injector = Guice.createInjector(Modules.override(Modules.override(new MainModule())
+                                                                         .with(new TestModule()))
+                                                        .with(binder -> binder.bind(LoggerHandler.class)
+                                                                              .toInstance(loggerHandler)));
         vertx = injector.getInstance(Vertx.class);
         config = injector.getInstance(Configuration.class);
         Router router = injector.getInstance(Router.class);
 
-        router.get("/500-route").handler(p -> {
-            throw new RuntimeException("Fails with 500");
-        });
+        router.get("/500-route").handler(p ->
+                                         {
+                                             throw new RuntimeException("Fails with 500");
+                                         });
 
-        router.get("/404-route").handler(p -> {
-            throw new HttpException(NOT_FOUND.code(), "Sorry, it's not here");
-        });
+        router.get("/404-route").handler(p ->
+                                         {
+                                             throw new HttpException(NOT_FOUND.code(), "Sorry, it's not here");
+                                         });
 
-        router.get("/204-route").handler(p -> {
-            throw new HttpException(NO_CONTENT.code(), "Sorry, no content");
-        });
+        router.get("/204-route").handler(p ->
+                                         {
+                                             throw new HttpException(NO_CONTENT.code(), "Sorry, no content");
+                                         });
 
         VertxTestContext context = new VertxTestContext();
-        HttpServer server = injector.getInstance(HttpServer.class);
+        server = injector.getInstance(HttpServer.class);
         server.listen(config.getPort(), context.succeedingThenComplete());
 
         context.awaitCompletion(5, TimeUnit.SECONDS);
     }
 
+    @AfterEach
+    void tearDown() throws InterruptedException
+    {
+        final CountDownLatch closeLatch = new CountDownLatch(1);
+        server.close(res -> closeLatch.countDown());
+        vertx.close();
+        if (closeLatch.await(60, TimeUnit.SECONDS)) {
+            logger.info("Close event received before timeout.");
+        } else {
+            logger.error("Close event timed out.");
+        }
+    }
+
     @DisplayName("Should log at error level when the request fails with a 500 code")
     @Test
     public void testInjectedLoggerHandlerLogsAtErrorLevel(VertxTestContext testContext)
     {
-        helper("/500-route", testContext, 500, "Internal Server Error");
+        helper("/500-route", testContext, INTERNAL_SERVER_ERROR.code(),
+               "Error 500: Internal Server Error");
     }
 
     @DisplayName("Should log at warn level when the request fails with a 404 error")
     @Test
     public void testInjectedLoggerHandlerLogsAtWarnLevel(VertxTestContext testContext)
     {
-        helper("/404-route", testContext, 404, "Not Found");
+        helper("/404-route", testContext, NOT_FOUND.code(), "Error 404: Not Found");
     }
 
-    @DisplayName("Should log at info level when the request returns with a 500 error")
+    @DisplayName("Should log at info level when the request returns with a 204 status code")
     @Test
     public void testInjectedLoggerHandlerLogsAtInfoLevel(VertxTestContext testContext)
     {
-        helper("/204-route", testContext, 204, null);
+        helper("/204-route", testContext, NO_CONTENT.code(), null);
     }
 
     private void helper(String requestURI, VertxTestContext testContext, int expectedStatusCode, String expectedBody)
     {
         WebClient client = WebClient.create(vertx);
-        Handler<HttpResponse<String>> responseVerifier = response -> testContext.verify(
-        () -> {
+        Handler<HttpResponse<String>> responseVerifier = response -> testContext.verify(() ->
+        {
             assertThat(response.statusCode()).isEqualTo(expectedStatusCode);
             if (expectedBody == null)
             {
@@ -110,8 +136,7 @@ private void helper(String requestURI, VertxTestContext testContext, int expecte
             verify(logger, times(1)).info("{}", expectedStatusCode);
         });
         client.get(config.getPort(), "localhost", requestURI)
-              .as(BodyCodec.string())
-              .ssl(false)
+              .as(BodyCodec.string()).ssl(false)
               .send(testContext.succeeding(responseVerifier));
     }
 

src/test/java/org/apache/cassandra/sidecar/StreamSSTableComponentTest.java

@@ -20,7 +20,13 @@
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 
-import static org.junit.Assert.assertEquals;
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
+import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.PARTIAL_CONTENT;
+import static io.netty.handler.codec.http.HttpResponseStatus.REQUESTED_RANGE_NOT_SATISFIABLE;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * Test for StreamSSTableComponent
@@ -42,7 +48,7 @@ void setUp() throws InterruptedException
         config = injector.getInstance(Configuration.class);
 
         VertxTestContext context = new VertxTestContext();
-        server.listen(config.getPort(), context.completing());
+        server.listen(config.getPort(), context.succeedingThenComplete());
 
         context.awaitCompletion(5, TimeUnit.SECONDS);
     }
@@ -69,8 +75,8 @@ void testRoute(VertxTestContext context)
                 .as(BodyCodec.buffer())
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(200, response.statusCode());
-                    assertEquals("data", response.bodyAsString());
+                    assertThat(response.statusCode()).isEqualTo(OK.code());
+                    assertThat(response.bodyAsString()).isEqualTo("data");
                     context.completeNow();
                 })));
     }
@@ -84,7 +90,7 @@ void testKeyspaceNotFound(VertxTestContext context)
         client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(404, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(NOT_FOUND.code());
                     context.completeNow();
                 })));
     }
@@ -98,7 +104,7 @@ void testSnapshotNotFound(VertxTestContext context)
         client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(404, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(NOT_FOUND.code());
                     context.completeNow();
                 })));
     }
@@ -112,8 +118,8 @@ void testForbiddenKeyspace(VertxTestContext context)
         client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(403, response.statusCode());
-                    assertEquals("system keyspace is forbidden", response.statusMessage());
+                    assertThat(response.statusCode()).isEqualTo(FORBIDDEN.code());
+                    assertThat(response.statusMessage()).isEqualTo("system keyspace is forbidden");
                     context.completeNow();
                 })));
     }
@@ -127,8 +133,8 @@ void testIncorrectKeyspaceFormat(VertxTestContext context)
         client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(400, response.statusCode());
-                    assertEquals("Invalid path params found", response.statusMessage());
+                    assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+                    assertThat(response.statusMessage()).isEqualTo("Invalid path params found");
                     context.completeNow();
                 })));
     }
@@ -142,8 +148,8 @@ void testIncorrectComponentFormat(VertxTestContext context)
         client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(400, response.statusCode());
-                    assertEquals("Invalid path params found", response.statusMessage());
+                    assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+                    assertThat(response.statusMessage()).isEqualTo("Invalid path params found");
                     context.completeNow();
                 })));
     }
@@ -157,8 +163,8 @@ void testAccessDeniedToCertainComponents(VertxTestContext context)
         client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(400, response.statusCode());
-                    assertEquals("Invalid path params found", response.statusMessage());
+                    assertThat(response.statusCode()).isEqualTo(BAD_REQUEST.code());
+                    assertThat(response.statusMessage()).isEqualTo("Invalid path params found");
                     context.completeNow();
                 })));
     }
@@ -174,8 +180,8 @@ void testPartialTableName(VertxTestContext context)
                 .as(BodyCodec.buffer())
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(200, response.statusCode());
-                    assertEquals("data", response.bodyAsString());
+                    assertThat(response.statusCode()).isEqualTo(OK.code());
+                    assertThat(response.bodyAsString()).isEqualTo("data");
                     context.completeNow();
                 })));
     }
@@ -190,7 +196,7 @@ void testInvalidRange(VertxTestContext context)
                 .putHeader("Range", "bytes=4-3")
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(416, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(REQUESTED_RANGE_NOT_SATISFIABLE.code());
                     context.completeNow();
                 })));
     }
@@ -205,7 +211,7 @@ void testRangeExceeds(VertxTestContext context)
                 .putHeader("Range", "bytes=5-9")
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(416, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(REQUESTED_RANGE_NOT_SATISFIABLE.code());
                     context.completeNow();
                 })));
     }
@@ -220,7 +226,7 @@ void testPartialRangeExceeds(VertxTestContext context)
                 .putHeader("Range", "bytes=5-")
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(416, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(REQUESTED_RANGE_NOT_SATISFIABLE.code());
                     context.completeNow();
                 })));
     }
@@ -236,8 +242,8 @@ void testRangeBoundaryExceeds(VertxTestContext context)
                 .as(BodyCodec.buffer())
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(200, response.statusCode());
-                    assertEquals("data", response.bodyAsString());
+                    assertThat(response.statusCode()).isEqualTo(OK.code());
+                    assertThat(response.bodyAsString()).isEqualTo("data");
                     context.completeNow();
                 })));
     }
@@ -253,8 +259,8 @@ void testPartialRangeStreamed(VertxTestContext context)
                 .as(BodyCodec.buffer())
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(206, response.statusCode());
-                    assertEquals("dat", response.bodyAsString());
+                    assertThat(response.statusCode()).isEqualTo(PARTIAL_CONTENT.code());
+                    assertThat(response.bodyAsString()).isEqualTo("dat");
                     context.completeNow();
                 })));
     }
@@ -270,8 +276,8 @@ void testSuffixRange(VertxTestContext context)
                 .as(BodyCodec.buffer())
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(206, response.statusCode());
-                    assertEquals("ta", response.bodyAsString());
+                    assertThat(response.statusCode()).isEqualTo(PARTIAL_CONTENT.code());
+                    assertThat(response.bodyAsString()).isEqualTo("ta");
                     context.completeNow();
                 })));
     }
@@ -286,7 +292,7 @@ void testSuffixRangeExceeds(VertxTestContext context)
                 .putHeader("Range", "bytes=-5")
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(416, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(REQUESTED_RANGE_NOT_SATISFIABLE.code());
                     context.completeNow();
                 })));
     }
@@ -301,7 +307,7 @@ void testInvalidRangeUnit(VertxTestContext context)
                 .putHeader("Range", "bits=0-2")
                 .send(context.succeeding(response -> context.verify(() ->
                 {
-                    assertEquals(416, response.statusCode());
+                    assertThat(response.statusCode()).isEqualTo(REQUESTED_RANGE_NOT_SATISFIABLE.code());
                     context.completeNow();
                 })));
     }
@@ -317,8 +323,8 @@ void testStreamingFromSpecificInstance(VertxTestContext context)
               .as(BodyCodec.buffer())
               .send(context.succeeding(response -> context.verify(() ->
               {
-                  assertEquals(200, response.statusCode());
-                  assertEquals("data", response.bodyAsString());
+                  assertThat(response.statusCode()).isEqualTo(OK.code());
+                  assertThat(response.bodyAsString()).isEqualTo("data");
                   context.completeNow();
               })));
     }