add log

Fungx · Dec 9, 2023 · beb2bc9 · beb2bc9
1 parent a8dd5a3
commit beb2bc9
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 5 deletions.
diff --git a/...c/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java b/...c/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java
@@ -84,10 +84,19 @@ private void doInit() throws Exception {
                     .map(MessageReader::toEvent)
                     .onSuccess(event -> {
                         queue.add(event);
+                        log.info("[HttpSourceConnector] Succeed to convert payload into CloudEvent. StatusCode={}", HttpResponseStatus.OK.code());
                         ctx.response().setStatusCode(HttpResponseStatus.OK.code()).end();
                     })
                     .onFailure(t -> {
-                        ctx.response().setStatusCode(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()).setStatusMessage(t.getMessage()).end();
+                        int code;
+                        if (t instanceof IllegalStateException) {
+                            code = HttpResponseStatus.BAD_REQUEST.code();
+                            log.error("[HttpSourceConnector] Malformed request. StatusCode={}", code, t);
+                        } else {
+                            code = HttpResponseStatus.INTERNAL_SERVER_ERROR.code();
+                            log.error("[HttpSourceConnector] Failed to convert payload into CloudEvent. StatusCode={}", code, t);
+                        }
+                        ctx.response().setStatusCode(code).setStatusMessage(t.getMessage()).end();
                     });
             });
         this.server = vertx.createHttpServer(new HttpServerOptions()
@@ -100,8 +109,8 @@ private void doInit() throws Exception {
     public void start() throws Exception {
         Throwable t = this.server.listen().cause();
         if (t != null) {
-            log.error("[HttpSourceConnector] failed to start Vertx server.", t);
-            throw new EventMeshException("failed to start Vertx server.");
+            log.error("[HttpSourceConnector] Failed to start Vertx server.", t);
+            throw new EventMeshException("Failed to start Vertx server");
         }
     }
 
@@ -119,8 +128,8 @@ public String name() {
     public void stop() throws Exception {
         Throwable t = this.server.close().cause();
         if (t != null) {
-            log.error("[HttpSourceConnector] failed to stop Vertx server.", t);
-            throw new EventMeshException("failed to stop Vertx server.");
+            log.error("[HttpSourceConnector] Failed to stop Vertx server.", t);
+            throw new EventMeshException("Failed to stop Vertx server");
         }
     }
 

diff --git a/...st/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnectorTest.java b/...st/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnectorTest.java
@@ -86,6 +86,13 @@ void testPoll() throws Exception {
         for (ConnectRecord r : res) {
             Assertions.assertEquals(expectedMessage, new String((byte[]) r.getData()));
         }
+
+        // test invalid requests
+        HttpPost invalidPost = new HttpPost(uri);
+        invalidPost.setHeader(HttpHeaders.CONTENT_TYPE, "text/plain");
+        invalidPost.setHeader("ce-id", String.valueOf(UUID.randomUUID()));
+        HttpResponse resp = httpClient.execute(invalidPost);
+        Assertions.assertNotEquals(HttpStatus.SC_OK, resp.getStatusLine().getStatusCode());
     }
 
     HttpResponse mockBinaryRequest() throws Exception {