upgrade opensearch sink to client 2.16 and tests to use server 2.16.0

Author: nikhil-ctds

pom.xml

@@ -158,7 +158,7 @@ flexible messaging model and an intuitive client API.</description>
     <bouncycastle.version>1.78.1</bouncycastle.version>
     <bouncycastle.bcpkix-fips.version>1.0.7</bouncycastle.bcpkix-fips.version>
     <bouncycastle.bc-fips.version>1.0.2.5</bouncycastle.bc-fips.version>
-    <jackson.version>2.14.2</jackson.version>
+    <jackson.version>2.16.0</jackson.version>
     <reflections.version>0.10.2</reflections.version>
     <swagger.version>1.6.2</swagger.version>
     <puppycrawl.checkstyle.version>10.14.2</puppycrawl.checkstyle.version>
@@ -194,7 +194,7 @@ flexible messaging model and an intuitive client API.</description>
     <openmldb-jdbc.version>0.4.4-hotfix1</openmldb-jdbc.version>
     <hdfs-offload-version3>3.3.5</hdfs-offload-version3>
     <json-smart.version>2.4.10</json-smart.version>
-    <opensearch.version>1.2.4</opensearch.version>
+    <opensearch.version>2.16.0</opensearch.version>
     <elasticsearch-java.version>8.12.1</elasticsearch-java.version>
     <trino.version>368</trino.version>
     <debezium.version>2.6.1.Final</debezium.version>

pulsar-common/src/main/java/org/apache/pulsar/common/util/FieldParser.java

@@ -21,8 +21,8 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
-import com.fasterxml.jackson.databind.AnnotationIntrospector;
-import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
+import com.fasterxml.jackson.databind.DeserializationConfig;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.util.EnumResolver;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
@@ -58,7 +58,7 @@ public final class FieldParser {
     private static final Map<String, Method> CONVERTERS = new HashMap<>();
     private static final Map<Class<?>, Class<?>> WRAPPER_TYPES = new HashMap<>();
 
-    private static final AnnotationIntrospector ANNOTATION_INTROSPECTOR = new JacksonAnnotationIntrospector();
+    private static final DeserializationConfig DESERIALIZATION_CONFIG = new ObjectMapper().getDeserializationConfig();
 
     static {
         // Preload converters and wrapperTypes.
@@ -100,7 +100,7 @@ public static <T> T convert(Object from, Class<T> to) {
 
         if (to.isEnum()) {
             // Converting string to enum
-            EnumResolver r = EnumResolver.constructUsingToString((Class<Enum<?>>) to, ANNOTATION_INTROSPECTOR);
+            EnumResolver r = EnumResolver.constructUsingToString(DESERIALIZATION_CONFIG, to);
             T value = (T) r.findEnum((String) from);
             if (value == null) {
                 throw new RuntimeException("Invalid value '" + from + "' for enum " + to);

pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java

@@ -49,12 +49,12 @@
 import org.opensearch.client.indices.CreateIndexRequest;
 import org.opensearch.client.indices.CreateIndexResponse;
 import org.opensearch.client.indices.GetIndexRequest;
-import org.opensearch.common.Strings;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.unit.ByteSizeUnit;
-import org.opensearch.common.unit.ByteSizeValue;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.xcontent.XContentType;
+import org.opensearch.core.common.Strings;
+import org.opensearch.core.common.unit.ByteSizeUnit;
+import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.search.builder.SearchSourceBuilder;
@@ -229,7 +229,7 @@ public boolean indexDocument(String index, String documentId, String documentSou
         if (!Strings.isNullOrEmpty(documentId)) {
             indexRequest.id(documentId);
         }
-        indexRequest.type(config.getTypeName());
+
         indexRequest.source(documentSource, XContentType.JSON);
 
         IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
@@ -245,7 +245,7 @@ public boolean indexDocument(String index, String documentId, String documentSou
     public boolean deleteDocument(String index, String documentId) throws IOException {
         DeleteRequest deleteRequest = Requests.deleteRequest(index);
         deleteRequest.id(documentId);
-        deleteRequest.type(config.getTypeName());
+
         DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
         if (log.isDebugEnabled()) {
             log.debug("delete result {}", deleteResponse.getResult());
@@ -301,7 +301,7 @@ public void appendIndexRequest(BulkProcessor.BulkIndexRequest request) throws IO
         if (!Strings.isNullOrEmpty(request.getDocumentId())) {
             indexRequest.id(request.getDocumentId());
         }
-        indexRequest.type(config.getTypeName());
+
         indexRequest.source(request.getDocumentSource(), XContentType.JSON);
         if (log.isDebugEnabled()) {
             log.debug("append index request id={}, type={}, source={}", request.getDocumentId(), config.getTypeName(),
@@ -314,7 +314,7 @@ public void appendIndexRequest(BulkProcessor.BulkIndexRequest request) throws IO
     public void appendDeleteRequest(BulkProcessor.BulkDeleteRequest request) throws IOException {
         DeleteRequest deleteRequest = new DeleteRequestWithPulsarRecord(request.getIndex(), request.getRecord());
         deleteRequest.id(request.getDocumentId());
-        deleteRequest.type(config.getTypeName());
+
         if (log.isDebugEnabled()) {
             log.debug("append delete request id={}, type={}", request.getDocumentId(), config.getTypeName());
         }

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchTestBase.java

@@ -46,7 +46,7 @@ public abstract class ElasticSearchTestBase {
             .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.17.7");
 
     public static final String OPENSEARCH = Optional.ofNullable(System.getenv("OPENSEARCH_IMAGE"))
-            .orElse("opensearchproject/opensearch:1.2.4");
+            .orElse("opensearchproject/opensearch:2.16.0");
 
     protected final String elasticImageName;
 
@@ -59,6 +59,7 @@ protected ElasticsearchContainer createElasticsearchContainer() {
         if (elasticImageName.equals(OPENSEARCH)) {
             DockerImageName dockerImageName = DockerImageName.parse(OPENSEARCH).asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch");
             elasticsearchContainer = new ElasticsearchContainer(dockerImageName)
+                    .withEnv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "0pEn7earch!")
                     .withEnv("OPENSEARCH_JAVA_OPTS", "-Xms128m -Xmx256m")
                     .withEnv("bootstrap.memory_lock", "true")
                     .withEnv("plugins.security.disabled", "true");

pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/opensearch/OpenSearchClientSslTests.java

@@ -77,7 +77,7 @@ public void testSslBasic() throws IOException {
                     .setElasticSearchUrl("https://" + container.getHttpHostAddress())
                     .setIndexName(INDEX)
                     .setUsername("admin")
-                    .setPassword("admin")
+                    .setPassword("0pEn7earch!")
                     .setSsl(new ElasticSearchSslConfig()
                             .setEnabled(true)
                             .setTruststorePath(sslResourceDir + "/truststore.jks")
@@ -102,7 +102,7 @@ public void testSslWithHostnameVerification() throws IOException {
                     .setElasticSearchUrl("https://" + container.getHttpHostAddress())
                     .setIndexName(INDEX)
                     .setUsername("admin")
-                    .setPassword("admin")
+                    .setPassword("0pEn7earch!")
                     .setSsl(new ElasticSearchSslConfig()
                             .setEnabled(true)
                             .setProtocols("TLSv1.2")
@@ -128,7 +128,7 @@ public void testSslWithClientAuth() throws IOException {
                     .setElasticSearchUrl("https://" + container.getHttpHostAddress())
                     .setIndexName(INDEX)
                     .setUsername("admin")
-                    .setPassword("admin")
+                    .setPassword("0pEn7earch!")
                     .setSsl(new ElasticSearchSslConfig()
                             .setEnabled(true)
                             .setHostnameVerification(true)

tests/docker-images/latest-version-image/conf/functions_worker.conf

@@ -22,7 +22,7 @@ autostart=false
 redirect_stderr=true
 stdout_logfile=/var/log/pulsar/functions_worker.log
 directory=/pulsar
-environment=PULSAR_MEM="-Xmx128M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/pulsar/logs/functions",PULSAR_GC="-XX:+UseZGC"
+environment=PULSAR_MEM="-Xmx128M -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/pulsar/logs/functions",PULSAR_GC="-XX:+UseG1GC"
 command=/pulsar/bin/pulsar functions-worker
 user=pulsar
 stopwaitsecs=15

tests/integration/pom.xml

@@ -220,6 +220,12 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+      <scope>test</scope>
+    </dependency>
+
     <!-- mongodb -->
     <dependency>
       <groupId>org.testcontainers</groupId>

tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java

@@ -35,7 +35,7 @@
 public class OpenSearchSinkTester extends ElasticSearchSinkTester {
 
     public static final String OPENSEARCH = Optional.ofNullable(System.getenv("OPENSEARCH_IMAGE"))
-            .orElse("opensearchproject/opensearch:1.2.4");
+            .orElse("opensearchproject/opensearch:2.16.0");
 
     private RestHighLevelClient elasticClient;
 
@@ -49,6 +49,7 @@ protected ElasticsearchContainer createElasticContainer() {
         DockerImageName dockerImageName = DockerImageName.parse(OPENSEARCH)
                 .asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch");
         return new ElasticsearchContainer(dockerImageName)
+                .withEnv("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "0pEn7earch!")
                 .withEnv("OPENSEARCH_JAVA_OPTS", "-Xms128m -Xmx256m")
                 .withEnv("bootstrap.memory_lock", "true")
                 .withEnv("plugins.security.disabled", "true");