fix: Revert "chore: upgrade vertx to 4.2 (#8975)" (#9003)

This reverts commit f5a4f60.
confluentinc · Apr 8, 2022 · d4214de · d4214de
1 parent 57f3596
commit d4214de
Show file tree

Hide file tree

Showing 40 changed files with 856 additions and 562 deletions.
diff --git a/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientImpl.java b/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientImpl.java
@@ -52,7 +52,6 @@
 import io.vertx.core.http.HttpClientResponse;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpVersion;
-import io.vertx.core.http.RequestOptions;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.JksOptions;
@@ -605,44 +604,26 @@ private <T extends CompletableFuture<?>> void makeRequest(
       final Handler<HttpClientResponse> responseHandler,
       final boolean endRequest,
       final HttpMethod method) {
-    final RequestOptions options = new RequestOptions();
-    options.setMethod(method);
-    options.setServer(serverSocketAddress);
-    options.setPort(clientOptions.getPort());
-    options.setHost(clientOptions.getHost());
-    options.setURI(path);
-
-    httpClient.request(options, ar -> {
-      if (ar.failed()) {
-        cf.completeExceptionally(ar.cause());
-      }
-
-      HttpClientRequest request = ar.result();
-      request.response(response -> {
-        if (response.failed()) {
-          cf.completeExceptionally(response.cause());
-        }
-
-        responseHandler.handle(response.result());
-      });
-      request.exceptionHandler(cf::completeExceptionally);
-
-      request = configureUserAgent(request);
-      if (clientOptions.isUseBasicAuth()) {
-        request = configureBasicAuth(request);
-      }
-      if (clientOptions.getRequestHeaders() != null) {
-        for (final Entry<String, String> entry : clientOptions.getRequestHeaders().entrySet()) {
-          request.putHeader(entry.getKey(), entry.getValue());
-        }
-      }
-      if (endRequest) {
-        request.end(requestBody);
-      } else {
-        final HttpClientRequest finalRequest = request;
-        finalRequest.sendHead(version -> finalRequest.writeCustomFrame(0, 0, requestBody));
+    HttpClientRequest request = httpClient.request(method,
+        serverSocketAddress, clientOptions.getPort(), clientOptions.getHost(),
+        path,
+        responseHandler)
+        .exceptionHandler(cf::completeExceptionally);
+    request = configureUserAgent(request);
+    if (clientOptions.isUseBasicAuth()) {
+      request = configureBasicAuth(request);
+    }
+    if (clientOptions.getRequestHeaders() != null) {
+      for (final Entry<String, String> entry : clientOptions.getRequestHeaders().entrySet()) {
+        request.putHeader(entry.getKey(), entry.getValue());
       }
-    });
+    }
+    if (endRequest) {
+      request.end(requestBody);
+    } else {
+      final HttpClientRequest finalRequest = request;
+      finalRequest.sendHead(version -> finalRequest.writeCustomFrame(0, 0, requestBody));
+    }
   }
 
   private HttpClientRequest configureBasicAuth(final HttpClientRequest request) {

diff --git a/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/QueryResponseHandler.java b/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/QueryResponseHandler.java
@@ -15,7 +15,8 @@
 
 package io.confluent.ksql.api.client.impl;
 
-import io.confluent.ksql.rest.ApiJsonMapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.confluent.ksql.api.client.util.JsonMapper;
 import io.confluent.ksql.rest.entity.QueryResponseMetadata;
 import io.vertx.core.Context;
 import io.vertx.core.buffer.Buffer;
@@ -24,6 +25,8 @@
 
 abstract class QueryResponseHandler<T extends CompletableFuture<?>> extends ResponseHandler<T> {
 
+  private static final ObjectMapper JSON_MAPPER = JsonMapper.get();
+
   protected boolean hasReadArguments;
 
   QueryResponseHandler(final Context context, final RecordParser recordParser, final T cf) {
@@ -59,8 +62,7 @@ private void handleArgs(final Buffer buff) {
 
     final QueryResponseMetadata queryResponseMetadata;
     try {
-      queryResponseMetadata = ApiJsonMapper.INSTANCE.get()
-          .readValue(buff.getBytes(), QueryResponseMetadata.class);
+      queryResponseMetadata = JSON_MAPPER.readValue(buff.getBytes(), QueryResponseMetadata.class);
     } catch (Exception e) {
       cf.completeExceptionally(e);
       return;

diff --git a/...pi-client/src/main/java/io/confluent/ksql/api/client/impl/StreamQueryResponseHandler.java b/...pi-client/src/main/java/io/confluent/ksql/api/client/impl/StreamQueryResponseHandler.java
@@ -45,9 +45,7 @@ public class StreamQueryResponseHandler
   private Map<String, Object> properties;
   private ClientImpl client;
 
-  StreamQueryResponseHandler(
-      final Context context,
-      final RecordParser recordParser,
+  StreamQueryResponseHandler(final Context context, final RecordParser recordParser,
       final CompletableFuture<StreamedQueryResult> cf,
       final AtomicReference<String> serializedCV,
       final AtomicReference<String> continuationToken,

diff --git a/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/util/JsonMapper.java b/ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/util/JsonMapper.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.client.util;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+
+public final class JsonMapper {
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  static {
+    MAPPER.disable(JsonGenerator.Feature.AUTO_CLOSE_TARGET)
+        .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+        .disable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES)
+        .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)
+        .enable(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN)
+        .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true))
+        .registerModule(new Jdk8Module());
+  }
+
+  private JsonMapper() {
+
+  }
+
+  public static ObjectMapper get() {
+    return MAPPER.copy();
+  }
+
+}
diff --git a/ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/impl/ClientImplTest.java b/ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/impl/ClientImplTest.java
@@ -20,19 +20,14 @@
 import static org.hamcrest.Matchers.is;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.when;
 
 import com.google.common.testing.EqualsTester;
 import io.confluent.ksql.api.client.Client;
 import io.confluent.ksql.api.client.ClientOptions;
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Future;
-import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.RequestOptions;
 import java.util.HashMap;
 import org.junit.Before;
 import org.junit.Test;
@@ -76,11 +71,7 @@ public void shouldSetUserAgent() {
 
     // When
     when(vertx.createHttpClient(any())).thenReturn(httpClient);
-    doAnswer(a -> {
-      ((Handler<AsyncResult<HttpClientRequest>>) a.getArgument(1))
-          .handle(Future.succeededFuture(clientRequest));
-      return null;
-    }).when(httpClient).request(any(RequestOptions.class), any(Handler.class));
+    when(httpClient.request(any(), any(), anyInt(), any(), any(), any())).thenReturn(clientRequest);
     when(clientRequest.exceptionHandler(any())).thenReturn(clientRequest);
     when(clientRequest.putHeader((String) any(), (String) any())).thenAnswer(a -> {
       String key = a.getArgument(0);

diff --git a/...-client/src/test/java/io/confluent/ksql/api/client/integration/ClientIntegrationTest.java b/...-client/src/test/java/io/confluent/ksql/api/client/integration/ClientIntegrationTest.java
@@ -37,7 +37,6 @@
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableListMultimap;
 import com.google.common.collect.ImmutableMap;
@@ -74,7 +73,6 @@
 import io.confluent.ksql.integration.IntegrationTestHarness;
 import io.confluent.ksql.integration.Retry;
 import io.confluent.ksql.name.ColumnName;
-import io.confluent.ksql.rest.ApiJsonMapper;
 import io.confluent.ksql.rest.entity.ConnectorList;
 import io.confluent.ksql.rest.entity.KsqlEntity;
 import io.confluent.ksql.rest.integration.RestIntegrationTestUtil;
@@ -90,11 +88,9 @@
 import io.confluent.ksql.serde.SerdeFeatures;
 import io.confluent.ksql.util.AppInfo;
 import io.confluent.ksql.util.KsqlConfig;
-import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.StructuredTypesDataProvider;
 import io.confluent.ksql.util.TestDataProvider;
 import io.vertx.core.Vertx;
-import io.vertx.core.json.Json;
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import java.io.FileOutputStream;
@@ -1341,7 +1337,7 @@ private static List<KsqlArray> convertToClientRows(
       }
 
       // Add header column
-      addObjectToKsqlArray(expectedRow, new byte[] {23});
+      expectedRow.add(new byte[] {23});
 
       expectedRows.add(expectedRow);
     }
@@ -1361,33 +1357,11 @@ private static void addObjectToKsqlArray(final KsqlArray array, final Object val
       array.add(SqlTimeTypes.formatDate((Date) value));
     } else if (value instanceof Time) {
       array.add(SqlTimeTypes.formatTime((Time) value));
-    } else if (value instanceof byte[]) {
-      array.add(serializeVertX3CompatibleByte((byte[]) value));
     } else {
       array.add(value);
     }
   }
 
-  /**
-   * VertX 4 changed to serialize using Base64 without padding but our server
-   * still encodes the data with padding - this uses the same serializer that
-   * the server uses to serialize the data into a string. Note that this issue
-   * only affects tests because the test uses string comparison to compare the
-   * bytes (it's schemaless, so the byte[] comes back as a string from the server).
-   *
-   * @see <a href=https://github.com/eclipse-vertx/vert.x/pull/3197>vertx#3197</a>
-   */
-  private static String serializeVertX3CompatibleByte(final byte[] bytes) {
-    try {
-      // writeValueAsString by default adds quotes to both sides to make it valid
-      // JSON
-      final String escaped = ApiJsonMapper.INSTANCE.get().writeValueAsString(bytes);
-      return escaped.substring(1, escaped.length() - 1);
-    } catch (JsonProcessingException e) {
-      throw new KsqlException(e);
-    }
-  }
-
   private static Matcher<? super StreamInfo> streamForProvider(
       final TestDataProvider testDataProvider
   ) {

diff --git a/...st/java/io/confluent/ksql/api/client/integration/PushV2ClientContinueIntegrationTest.java b/...st/java/io/confluent/ksql/api/client/integration/PushV2ClientContinueIntegrationTest.java
@@ -31,7 +31,6 @@
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Multimap;
 import io.confluent.common.utils.IntegrationTest;
@@ -49,7 +48,6 @@
 import io.confluent.ksql.api.client.util.RowUtil;
 import io.confluent.ksql.integration.IntegrationTestHarness;
 import io.confluent.ksql.integration.Retry;
-import io.confluent.ksql.rest.ApiJsonMapper;
 import io.confluent.ksql.rest.entity.KsqlEntity;
 import io.confluent.ksql.rest.integration.RestIntegrationTestUtil;
 import io.confluent.ksql.rest.server.KsqlRestConfig;
@@ -58,7 +56,6 @@
 import io.confluent.ksql.serde.Format;
 import io.confluent.ksql.serde.FormatFactory;
 import io.confluent.ksql.util.KsqlConfig;
-import io.confluent.ksql.util.KsqlException;
 import io.confluent.ksql.util.PersistentQueryMetadata;
 import io.confluent.ksql.util.StructuredTypesDataProvider;
 import io.confluent.ksql.util.StructuredTypesDataProvider.Batch;
@@ -363,7 +360,7 @@ private static List<KsqlArray> convertToClientRows(
       }
 
       // Add header column
-      addObjectToKsqlArray(expectedRow, new byte[] {23});
+      expectedRow.add(new byte[] {23});
 
       expectedRows.add(expectedRow);
     }
@@ -383,31 +380,11 @@ private static void addObjectToKsqlArray(final KsqlArray array, final Object val
       array.add(SqlTimeTypes.formatDate((Date) value));
     } else if (value instanceof Time) {
       array.add(SqlTimeTypes.formatTime((Time) value));
-    } else if (value instanceof byte[]) {
-      array.add(serializeVertX3CompatibleByte((byte[]) value));
     } else {
       array.add(value);
     }
   }
 
-  /**
-   * VertX 4 changed to serialize using Base64 without padding but our server
-   * still encodes the data with padding - this uses the same serializer that
-   * the server uses to serialize the data into a string
-   *
-   * @see <a href=https://github.com/eclipse-vertx/vert.x/pull/3197>vertx#3197</a>
-   */
-  private static String serializeVertX3CompatibleByte(final byte[] bytes) {
-    try {
-      // writeValueAsString by default adds quotes to both sides to make it valid
-      // JSON
-      final String escaped = ApiJsonMapper.INSTANCE.get().writeValueAsString(bytes);
-      return escaped.substring(1, escaped.length() - 1);
-    } catch (JsonProcessingException e) {
-      throw new KsqlException(e);
-    }
-  }
-
   private void assertExpectedScalablePushQueries(
       final int expectedScalablePushQueries
   ) {

diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/reactive/BasePublisher.java b/ksqldb-common/src/main/java/io/confluent/ksql/reactive/BasePublisher.java
@@ -18,7 +18,6 @@
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.confluent.ksql.util.VertxUtils;
 import io.vertx.core.Context;
-import io.vertx.core.Future;
 import java.util.Objects;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -66,9 +65,8 @@ public void subscribe(final Subscriber<? super T> subscriber) {
     }
   }
 
-  public Future<Void> close() {
+  public void close() {
     ctx.runOnContext(v -> doClose());
-    return Future.succeededFuture();
   }
 
   @SuppressFBWarnings(value = "EI_EXPOSE_REP", justification = "ctx should be mutable")

diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/util/VertxUtils.java b/ksqldb-common/src/main/java/io/confluent/ksql/util/VertxUtils.java
@@ -17,7 +17,6 @@
 
 import io.vertx.core.Context;
 import io.vertx.core.Vertx;
-import io.vertx.core.impl.ContextInternal;
 
 /**
  * General purpose utils (not limited to the server, could be used by client too) for the API
@@ -41,20 +40,7 @@ public static void checkContext(final Context context) {
   }
 
   public static boolean isEventLoopAndSameContext(final Context context) {
-    return Context.isOnEventLoopThread()
-        && (context == Vertx.currentContext()
-        || checkDuplicateContext(Vertx.currentContext(), context)
-        || checkDuplicateContext(context, Vertx.currentContext()));
-  }
-
-  private static boolean checkDuplicateContext(final Context context, final Context other) {
-    // see https://github.com/eclipse-vertx/vert.x/issues/3300 - the recommendation from
-    // the VertX community is to always call runOnContext() despite the performance overhead
-    // instead of checking the context. this hack allows us to keep the same pattern we had
-    // from before the VertX 4 migration
-    return context instanceof ContextInternal
-        && ((ContextInternal) context).isDuplicate()
-        && (((ContextInternal) context).unwrap() == other);
+    return Context.isOnEventLoopThread() && context == Vertx.currentContext();
   }
 
 }
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/execution/scalablepush/PushRouting.java b/ksqldb-engine/src/main/java/io/confluent/ksql/execution/scalablepush/PushRouting.java
@@ -291,8 +291,8 @@ static CompletableFuture<RoutingResult> executeOrRouteQuery(
       final String thisHostName
   ) {
     if (node.isLocal()) {
-      LOG.info("Query with id {} executed locally at host {} at timestamp {}.",
-          pushPhysicalPlanManager.getQueryId(), node.location(),
+      LOG.info("Query {} id {} executed locally at host {} at timestamp {}.",
+          statement.getStatementText(), pushPhysicalPlanManager.getQueryId(), node.location(),
           System.currentTimeMillis());
       scalablePushQueryMetrics
           .ifPresent(metrics -> metrics.recordLocalRequests(1));