test: update HttpSinkConnectorTest

Author: cnzakii

Diff for: eventmesh-connectors/eventmesh-connector-http/build.gradle

@@ -26,6 +26,7 @@ dependencies {
 
     testImplementation "org.apache.httpcomponents:httpclient"
     testImplementation 'org.mock-server:mockserver-netty:5.15.0'
+    testImplementation 'com.squareup.okhttp3:okhttp:4.12.0'
     compileOnly 'org.projectlombok:lombok'
     annotationProcessor 'org.projectlombok:lombok'
 }

Diff for: eventmesh-connectors/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/source/connector/HttpSinkConnectorTest.java

@@ -21,6 +21,7 @@
 
 import org.apache.eventmesh.connector.http.sink.HttpSinkConnector;
 import org.apache.eventmesh.connector.http.sink.config.HttpSinkConfig;
+import org.apache.eventmesh.connector.http.sink.config.HttpWebhookConfig;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
 import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
@@ -34,18 +35,21 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.mockserver.client.MockServerClient;
 import org.mockserver.integration.ClientAndServer;
 import org.mockserver.model.HttpRequest;
 import org.mockserver.model.HttpResponse;
 import org.mockserver.model.MediaType;
-import org.mockserver.verify.VerificationTimes;
-
-import io.vertx.core.http.HttpMethod;
 
 import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 
+import okhttp3.HttpUrl;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+
 public class HttpSinkConnectorTest {
 
     private HttpSinkConnector sinkConnector;
@@ -68,8 +72,7 @@ void before() throws Exception {
         this.severUri = URI.create(sinkConfig.connectorConfig.getUrls()[0]);
         // start mockServer
         mockServer = ClientAndServer.startClientAndServer(severUri.getPort());
-        // mockServer response
-        new MockServerClient(severUri.getHost(), severUri.getPort())
+        mockServer.reset()
             .when(
                 request()
                     .withMethod("POST")
@@ -113,46 +116,49 @@ void testPut() throws Exception {
         Thread.sleep(5000);
 
         // verify request
-        new MockServerClient(severUri.getHost(), severUri.getPort())
-            .verify(
-                HttpRequest.request()
-                    .withMethod(HttpMethod.POST.name())
-                    .withPath(severUri.getPath()),
-                VerificationTimes.exactly(times));
-
-        /*
-        **The following code is only required in webhook mode**
+        HttpRequest[] recordedRequests = mockServer.retrieveRecordedRequests(null);
+        assert recordedRequests.length == times;
 
         // verify response
         HttpWebhookConfig webhookConfig = sinkConfig.connectorConfig.getWebhookConfig();
-        URI uri = new URIBuilder()
-            .setScheme("http")
-            .setHost(severUri.getHost())
-            .setPort(webhookConfig.getPort())
-            .setPath(webhookConfig.getExportPath())
-            .addParameter("pageNum", "1")
-            .addParameter("pageSize", "10")
-            .addParameter("type", "poll")
+        String url = new HttpUrl.Builder()
+            .scheme("http")
+            .host(severUri.getHost())
+            .port(webhookConfig.getPort())
+            .addPathSegments(webhookConfig.getExportPath())
+            .addQueryParameter("pageNum", "1")
+            .addQueryParameter("pageSize", "10")
+            .addQueryParameter("type", "poll")
+            .build().toString();
+
+        // build request
+        Request request = new Request.Builder()
+            .url(url)
+            .addHeader("Content-Type", "application/json")
             .build();
 
-        CloseableHttpClient httpClient = HttpClients.createDefault();
-        HttpGet httpGet = new HttpGet(uri);
-        httpGet.setHeader("Content-Type", "application/json");
-        CloseableHttpResponse response = httpClient.execute(httpGet);
-        String body = EntityUtils.toString(response.getEntity());
-        assert body != null;
-        JSONArray pageItems = JSON.parseObject(body).getJSONArray("pageItems");
-        assert pageItems != null && pageItems.size() == times;
-        for (int i = 0; i < times; i++) {
-            JSONObject pageItem = pageItems.getJSONObject(i);
-            assert pageItem != null;
-            assert pageItem.getJSONObject("data") != null;
-            assert pageItem.getJSONObject("metadata") != null;
+        OkHttpClient client = new OkHttpClient();
+        try (Response response = client.newCall(request).execute()) {
+            // check response code
+            if (!response.isSuccessful()) {
+                throw new RuntimeException("Unexpected response code: " + response);
+            }
+            // check response body
+            ResponseBody responseBody = response.body();
+            if (responseBody != null) {
+                JSONObject jsonObject = JSON.parseObject(responseBody.string());
+                JSONArray pageItems = jsonObject.getJSONArray("pageItems");
+
+                assert pageItems != null && pageItems.size() == times;
+
+                for (int i = 0; i < times; i++) {
+                    JSONObject pageItem = pageItems.getJSONObject(i);
+                    assert pageItem != null;
+                    assert pageItem.getJSONObject("data") != null;
+                    assert pageItem.getJSONObject("metadata") != null;
+                }
+            }
         }
-
-        httpClient.close();
-
-         */
     }
 
     private ConnectRecord createConnectRecord() {

Diff for: eventmesh-connectors/eventmesh-connector-http/src/test/resources/sink-config.yml

@@ -39,7 +39,7 @@ connectorConfig:
     interval: 1000
     retryOnNonSuccess: true
   webhookConfig:
-    activate: false
+    activate: true
     exportPath: /export
     port: 8988
     serverIdleTimeout: 5000