trying to upgrade the client

datastax · Sep 10, 2024 · 2657eff · 2657eff
1 parent 3929079
commit 2657eff
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 8 deletions.
diff --git a/pom.xml b/pom.xml
@@ -159,7 +159,7 @@ flexible messaging model and an intuitive client API.</description>
     <mariadb-jdbc.version>2.7.5</mariadb-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.5.2</elasticsearch-java.version>
     <presto.version>334</presto.version>
     <scala.binary.version>2.13</scala.binary.version>

diff --git a/...a/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java b/...a/org/apache/pulsar/io/elasticsearch/client/opensearch/OpenSearchHighLevelRestClient.java
@@ -50,12 +50,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,8 @@ public boolean indexDocument(String index, String documentId, String documentSou
         if (!Strings.isNullOrEmpty(documentId)) {
             indexRequest.id(documentId);
         }
-        indexRequest.type(config.getTypeName());
+        // no longer needed?
+        //indexRequest.type(config.getTypeName());
         indexRequest.source(documentSource, XContentType.JSON);
 
         IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
@@ -245,7 +246,8 @@ 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());
+        // no longer needed?
+        //deleteRequest.type(config.getTypeName());
         DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
         if (log.isDebugEnabled()) {
             log.debug("delete result {}", deleteResponse.getResult());
@@ -301,7 +303,8 @@ public void appendIndexRequest(BulkProcessor.BulkIndexRequest request) throws IO
         if (!Strings.isNullOrEmpty(request.getDocumentId())) {
             indexRequest.id(request.getDocumentId());
         }
-        indexRequest.type(config.getTypeName());
+        // no longer needed?
+        //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 +317,8 @@ 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());
+        // no longer needed?
+        //deleteRequest.type(config.getTypeName());
         if (log.isDebugEnabled()) {
             log.debug("append delete request id={}, type={}", request.getDocumentId(), config.getTypeName());
         }

diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
@@ -207,6 +207,11 @@
       <artifactId>aws-java-sdk-core</artifactId>
       <scope>test</scope>
     </dependency>
+      <dependency>
+          <groupId>org.projectlombok</groupId>
+          <artifactId>lombok</artifactId>
+          <scope>test</scope>
+      </dependency>
 
   </dependencies>
 

diff --git a/...tion/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java b/...tion/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java
@@ -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");