fixing more tests

masseyke · Nov 30, 2023 · d3177f7 · d3177f7
1 parent c671543
commit d3177f7
Show file tree

Hide file tree

Showing 7 changed files with 83 additions and 70 deletions.
diff --git a/...plugin/core/src/main/java/org/elasticsearch/xpack/core/indexing/AsyncTwoPhaseIndexer.java b/...plugin/core/src/main/java/org/elasticsearch/xpack/core/indexing/AsyncTwoPhaseIndexer.java
@@ -557,6 +557,8 @@ private void onSearchResponse(SearchResponse searchResponse) {
                     }
                 } catch (Exception e) {
                     finishWithFailure(e);
+                } finally {
+                    bulkRequest.close();
                 }
             }
         } catch (Exception e) {

diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/termsenum/TermsEnumTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/termsenum/TermsEnumTests.java
@@ -201,16 +201,17 @@ public void testTermsEnumIPRandomized() throws Exception {
 
         // create random ip test data
         InetAddress[] randomIps = new InetAddress[numDocs];
-        BulkRequestBuilder bulkRequestBuilder = client().prepareBulk(indexName);
-        for (int i = 0; i < numDocs; i++) {
-            randomIps[i] = randomIp(randomBoolean());
-            bulkRequestBuilder.add(
-                prepareIndex(indexName).setSource(
-                    jsonBuilder().startObject().field("ip_addr", NetworkAddress.format(randomIps[i])).endObject()
-                )
-            );
+        try (BulkRequestBuilder bulkRequestBuilder = client().prepareBulk(indexName)) {
+            for (int i = 0; i < numDocs; i++) {
+                randomIps[i] = randomIp(randomBoolean());
+                bulkRequestBuilder.add(
+                    prepareIndex(indexName).setSource(
+                        jsonBuilder().startObject().field("ip_addr", NetworkAddress.format(randomIps[i])).endObject()
+                    )
+                );
+            }
+            assertNoFailures(bulkRequestBuilder.get());
         }
-        assertNoFailures(bulkRequestBuilder.get());
         assertAllSuccessful(client().admin().indices().prepareRefresh().get());
 
         // test for short random prefixes, max length 7 should at least include some separators but not be too long for short ipv4

diff --git a/...nalClusterTest/java/org/elasticsearch/xpack/downsample/DownsampleClusterDisruptionIT.java b/...nalClusterTest/java/org/elasticsearch/xpack/downsample/DownsampleClusterDisruptionIT.java
@@ -390,31 +390,32 @@ private void assertTargetIndex(final InternalTestCluster cluster, final String s
 
     private int bulkIndex(final String indexName, final DownsampleActionSingleNodeTests.SourceSupplier sourceSupplier, int docCount)
         throws IOException {
-        BulkRequestBuilder bulkRequestBuilder = internalCluster().client().prepareBulk();
-        bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
-        for (int i = 0; i < docCount; i++) {
-            IndexRequest indexRequest = new IndexRequest(indexName).opType(DocWriteRequest.OpType.CREATE);
-            XContentBuilder source = sourceSupplier.get();
-            indexRequest.source(source);
-            bulkRequestBuilder.add(indexRequest);
-        }
-        BulkResponse bulkResponse = bulkRequestBuilder.get();
-        int duplicates = 0;
-        for (BulkItemResponse response : bulkResponse.getItems()) {
-            if (response.isFailed()) {
-                if (response.getFailure().getCause() instanceof VersionConflictEngineException) {
-                    // A duplicate event was created by random generator. We should not fail for this
-                    // reason.
-                    logger.debug("We tried to insert a duplicate: [{}]", response.getFailureMessage());
-                    duplicates++;
-                } else {
-                    fail("Failed to index data: " + bulkResponse.buildFailureMessage());
+        try (BulkRequestBuilder bulkRequestBuilder = internalCluster().client().prepareBulk()) {
+            bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+            for (int i = 0; i < docCount; i++) {
+                IndexRequest indexRequest = new IndexRequest(indexName).opType(DocWriteRequest.OpType.CREATE);
+                XContentBuilder source = sourceSupplier.get();
+                indexRequest.source(source);
+                bulkRequestBuilder.add(indexRequest);
+            }
+            BulkResponse bulkResponse = bulkRequestBuilder.get();
+            int duplicates = 0;
+            for (BulkItemResponse response : bulkResponse.getItems()) {
+                if (response.isFailed()) {
+                    if (response.getFailure().getCause() instanceof VersionConflictEngineException) {
+                        // A duplicate event was created by random generator. We should not fail for this
+                        // reason.
+                        logger.debug("We tried to insert a duplicate: [{}]", response.getFailureMessage());
+                        duplicates++;
+                    } else {
+                        fail("Failed to index data: " + bulkResponse.buildFailureMessage());
+                    }
                 }
             }
+            int docsIndexed = docCount - duplicates;
+            logger.info("Indexed [{}] documents. Dropped [{}] duplicates.", docsIndexed, duplicates);
+            return docsIndexed;
         }
-        int docsIndexed = docCount - duplicates;
-        logger.info("Indexed [{}] documents. Dropped [{}] duplicates.", docsIndexed, duplicates);
-        return docsIndexed;
     }
 
     private void prepareSourceIndex(String sourceIndex) {

diff --git a/...lClusterTest/java/org/elasticsearch/xpack/ml/integration/CategorizeTextAggregationIT.java b/...lClusterTest/java/org/elasticsearch/xpack/ml/integration/CategorizeTextAggregationIT.java
@@ -100,43 +100,44 @@ private void createSourceData() {
 
         long nowMillis = System.currentTimeMillis();
 
-        BulkRequestBuilder bulkRequestBuilder = client().prepareBulk();
-        IndexRequest indexRequest = new IndexRequest(DATA_INDEX);
-        indexRequest.source("time", nowMillis - TimeValue.timeValueHours(2).millis(), "msg", "Node 1 started", "part", "nodes");
-        bulkRequestBuilder.add(indexRequest);
-        indexRequest = new IndexRequest(DATA_INDEX);
-        indexRequest.source(
-            "time",
-            nowMillis - TimeValue.timeValueHours(2).millis() + 1,
-            "msg",
-            "Failed to shutdown [error org.aaaa.bbbb.Cccc line 54 caused by foo exception]",
-            "part",
-            "shutdowns"
-        );
-        bulkRequestBuilder.add(indexRequest);
-        indexRequest = new IndexRequest(DATA_INDEX);
-        indexRequest.source(
-            "time",
-            nowMillis - TimeValue.timeValueHours(2).millis() + 1,
-            "msg",
-            "Failed to shutdown [error org.aaaa.bbbb.Cccc line 55 caused by foo exception]",
-            "part",
-            "shutdowns"
-        );
-        bulkRequestBuilder.add(indexRequest);
-        indexRequest = new IndexRequest(DATA_INDEX);
-        indexRequest.source("time", nowMillis - TimeValue.timeValueHours(1).millis(), "msg", "Node 2 started", "part", "nodes");
-        bulkRequestBuilder.add(indexRequest);
-        indexRequest = new IndexRequest(DATA_INDEX);
-        indexRequest.source("time", nowMillis, "msg", "Node 3 started", "part", "nodes");
-        bulkRequestBuilder.add(indexRequest);
-
-        indexRequest = new IndexRequest(DATA_INDEX);
-        indexRequest.source("time", nowMillis, "msg", "Node 3 stopped", "part", "nodes");
-        bulkRequestBuilder.add(indexRequest);
-
-        BulkResponse bulkResponse = bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
-        assertThat(bulkResponse.hasFailures(), is(false));
+        try (BulkRequestBuilder bulkRequestBuilder = client().prepareBulk()) {
+            IndexRequest indexRequest = new IndexRequest(DATA_INDEX);
+            indexRequest.source("time", nowMillis - TimeValue.timeValueHours(2).millis(), "msg", "Node 1 started", "part", "nodes");
+            bulkRequestBuilder.add(indexRequest);
+            indexRequest = new IndexRequest(DATA_INDEX);
+            indexRequest.source(
+                "time",
+                nowMillis - TimeValue.timeValueHours(2).millis() + 1,
+                "msg",
+                "Failed to shutdown [error org.aaaa.bbbb.Cccc line 54 caused by foo exception]",
+                "part",
+                "shutdowns"
+            );
+            bulkRequestBuilder.add(indexRequest);
+            indexRequest = new IndexRequest(DATA_INDEX);
+            indexRequest.source(
+                "time",
+                nowMillis - TimeValue.timeValueHours(2).millis() + 1,
+                "msg",
+                "Failed to shutdown [error org.aaaa.bbbb.Cccc line 55 caused by foo exception]",
+                "part",
+                "shutdowns"
+            );
+            bulkRequestBuilder.add(indexRequest);
+            indexRequest = new IndexRequest(DATA_INDEX);
+            indexRequest.source("time", nowMillis - TimeValue.timeValueHours(1).millis(), "msg", "Node 2 started", "part", "nodes");
+            bulkRequestBuilder.add(indexRequest);
+            indexRequest = new IndexRequest(DATA_INDEX);
+            indexRequest.source("time", nowMillis, "msg", "Node 3 started", "part", "nodes");
+            bulkRequestBuilder.add(indexRequest);
+
+            indexRequest = new IndexRequest(DATA_INDEX);
+            indexRequest.source("time", nowMillis, "msg", "Node 3 stopped", "part", "nodes");
+            bulkRequestBuilder.add(indexRequest);
+
+            BulkResponse bulkResponse = bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+            assertThat(bulkResponse.hasFailures(), is(false));
+        }
     }
 
 }
diff --git a/...lugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobResultsProvider.java b/...lugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobResultsProvider.java
@@ -1962,7 +1962,11 @@ public void removeJobFromCalendars(String jobId, ActionListener<Boolean> listene
                     ActionListener.releaseAfter(updateCalendarsListener, bulkUpdate)
                 );
             } else {
-                listener.onResponse(true);
+                try {
+                    listener.onResponse(true);
+                } finally {
+                    bulkUpdate.close();
+                }
             }
         }, listener::onFailure);
 

diff --git a/...ck/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/TokenService.java b/...ck/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/TokenService.java
@@ -927,6 +927,7 @@ private void indexInvalidation(
             logger.warn("No [{}] tokens provided for invalidation", srcPrefix);
             listener.onFailure(invalidGrantException("No tokens provided for invalidation"));
         } else {
+            if (true) System.out.println("**** client class: " + client.getClass() + ", " + client.prepareBulk());
             BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
             for (String tokenId : tokenIds) {
                 UpdateRequest request = client.prepareUpdate(tokensIndexManager.aliasName(), getTokenDocumentId(tokenId))

diff --git a/...ugin/security/src/test/java/org/elasticsearch/xpack/security/authc/TokenServiceTests.java b/...ugin/security/src/test/java/org/elasticsearch/xpack/security/authc/TokenServiceTests.java
@@ -16,8 +16,10 @@
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.NoShardAvailableActionException;
 import org.elasticsearch.action.UnavailableShardsException;
+import org.elasticsearch.action.bulk.BulkAction;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.get.GetAction;
 import org.elasticsearch.action.get.GetRequest;
@@ -166,6 +168,7 @@ public void setupClient() {
             return builder;
         }).when(client).prepareGet(anyString(), anyString());
         when(client.prepareIndex(any(String.class))).thenReturn(new IndexRequestBuilder(client, IndexAction.INSTANCE));
+        when(client.prepareBulk()).thenReturn(new BulkRequestBuilder(client, BulkAction.INSTANCE));
         when(client.prepareUpdate(any(String.class), any(String.class))).thenAnswer(inv -> {
             final String index = (String) inv.getArguments()[0];
             final String id = (String) inv.getArguments()[1];
@@ -443,7 +446,7 @@ public void testAuthnWithInvalidatedToken() throws Exception {
         TokenService tokenService = createTokenService(tokenServiceEnabledSettings, systemUTC());
         Authentication authentication = AuthenticationTestHelper.builder()
             .user(new User("joe", "admin"))
-            .realmRef(new RealmRef("native_realm", "native", "node1"))
+            .realmRef(new RealmRef("native_realm", "na7tive", "node1"))
             .build(false);
         PlainActionFuture<TokenService.CreateTokenResult> tokenFuture = new PlainActionFuture<>();
         Tuple<byte[], byte[]> newTokenBytes = tokenService.getRandomTokenBytes(randomBoolean());