You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by bo...@apache.org on 2022/11/11 04:19:06 UTC
[pulsar] branch branch-2.9 updated: [fix][flaky-test] ElasticSearchClientTests.testBulkBlocking (#16920)
This is an automated email from the ASF dual-hosted git repository.
bogong pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new aae0c69c5c6 [fix][flaky-test] ElasticSearchClientTests.testBulkBlocking (#16920)
aae0c69c5c6 is described below
commit aae0c69c5c6086b58cadc57fceea482693f95f33
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Tue Aug 9 14:07:31 2022 +0800
[fix][flaky-test] ElasticSearchClientTests.testBulkBlocking (#16920)
(cherry picked from commit f02679d83402ac07e06c48570505f343e350a491)
---
.../broker/auth/AuthorizationWithAuthDataTest.java | 3 +-
.../io/elasticsearch/ElasticSearchClientTests.java | 68 ++++++++++++----------
2 files changed, 40 insertions(+), 31 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
index ccb401f74b3..eb9ed006c13 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationWithAuthDataTest.java
@@ -25,6 +25,7 @@ import io.jsonwebtoken.SignatureAlgorithm;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
@@ -262,7 +263,7 @@ public class AuthorizationWithAuthDataTest extends MockedPulsarServiceBaseTest {
@Test
public void testAdmin() throws PulsarAdminException {
admin.tenants().createTenant("test-tenant-1",
- TenantInfo.builder().allowedClusters(Set.of(configClusterName)).build());
+ TenantInfo.builder().allowedClusters(Collections.singleton(configClusterName)).build());
admin.namespaces().createNamespace("test-tenant-1/test-namespace-1");
String partitionedTopic = UUID.randomUUID().toString();
admin.topics().createPartitionedTopic(partitionedTopic,3);
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java
index 1a505dae31a..1cf730f7b06 100644
--- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClientTests.java
@@ -31,11 +31,10 @@ import org.testcontainers.elasticsearch.ElasticsearchContainer;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
-
import java.io.IOException;
import java.util.Optional;
import java.util.UUID;
-
+import java.util.concurrent.atomic.LongAdder;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
@@ -63,8 +62,16 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
}
static class MockRecord<T> implements Record<T> {
- int acked = 0;
- int failed = 0;
+ LongAdder acked = new LongAdder();
+ LongAdder failed = new LongAdder();
+
+ public int getAcked() {
+ return acked.intValue();
+ }
+
+ public int getFailed() {
+ return failed.intValue();
+ }
@Override
public T getValue() {
@@ -73,12 +80,12 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
@Override
public void ack() {
- acked++;
+ acked.increment();
}
@Override
public void fail() {
- failed++;
+ failed.increment();
}
}
@@ -130,13 +137,13 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
try {
MockRecord<GenericObject> mockRecord = new MockRecord<>();
client.indexDocument(mockRecord, Pair.of("1", "{ \"a\":1}"));
- assertEquals(mockRecord.acked, 1);
- assertEquals(mockRecord.failed, 0);
+ assertEquals(mockRecord.getAcked(), 1);
+ assertEquals(mockRecord.getFailed(), 0);
assertEquals(client.totalHits(index), 1);
client.deleteDocument(mockRecord, "1");
- assertEquals(mockRecord.acked, 2);
- assertEquals(mockRecord.failed, 0);
+ assertEquals(mockRecord.getAcked(), 2);
+ assertEquals(mockRecord.getFailed(), 0);
assertEquals(client.totalHits(index), 0);
} finally {
client.delete(index);
@@ -193,11 +200,11 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
client.flush();
assertNotNull(client.irrecoverableError.get());
assertTrue(client.irrecoverableError.get().getMessage().contains("mapper_parsing_exception"));
- assertEquals(mockRecord.acked, 1);
- assertEquals(mockRecord.failed, 1);
+ assertEquals(mockRecord.getAcked(), 1);
+ assertEquals(mockRecord.getFailed(), 1);
assertThrows(Exception.class, () -> client.bulkIndex(mockRecord, Pair.of("3", "{\"a\":3}")));
- assertEquals(mockRecord.acked, 1);
- assertEquals(mockRecord.failed, 2);
+ assertEquals(mockRecord.getAcked(), 1);
+ assertEquals(mockRecord.getFailed(), 2);
}
}
@@ -215,8 +222,8 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
client.bulkIndex(mockRecord, Pair.of("2", "{\"a\":\"toto\"}"));
client.flush();
assertNull(client.irrecoverableError.get());
- assertEquals(mockRecord.acked, 1);
- assertEquals(mockRecord.failed, 1);
+ assertEquals(mockRecord.getAcked(), 1);
+ assertEquals(mockRecord.getFailed(), 1);
}
}
@@ -239,22 +246,22 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
MockRecord<GenericObject> mockRecord = new MockRecord<>();
client.bulkIndex(mockRecord, Pair.of("1", "{\"a\":1}"));
client.bulkIndex(mockRecord, Pair.of("2", "{\"a\":2}"));
- assertEquals(mockRecord.acked, 2);
- assertEquals(mockRecord.failed, 0);
+ assertEquals(mockRecord.getAcked(), 2);
+ assertEquals(mockRecord.getFailed(), 0);
assertEquals(client.totalHits(index), 2);
ChaosContainer<?> chaosContainer = ChaosContainer.pauseContainerForSeconds(container.getContainerName(), 15);
chaosContainer.start();
client.bulkIndex(mockRecord, Pair.of("3", "{\"a\":3}"));
- assertEquals(mockRecord.acked, 2);
- assertEquals(mockRecord.failed, 0);
+ assertEquals(mockRecord.getAcked(), 2);
+ assertEquals(mockRecord.getFailed(), 0);
assertEquals(client.totalHits(index), 2);
chaosContainer.stop();
client.flush();
- assertEquals(mockRecord.acked, 3);
- assertEquals(mockRecord.failed, 0);
+ assertEquals(mockRecord.getAcked(), 3);
+ assertEquals(mockRecord.getFailed(), 0);
assertEquals(client.totalHits(index), 3);
} finally {
client.delete(index);
@@ -284,14 +291,14 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
}
Awaitility.await().untilAsserted(() -> {
- assertThat("acked record", mockRecord.acked, greaterThanOrEqualTo(4));
- assertEquals(mockRecord.failed, 0);
+ assertThat("acked record", mockRecord.getAcked(), greaterThanOrEqualTo(4));
+ assertEquals(mockRecord.getFailed(), 0);
assertThat("totalHits", client.totalHits(index), greaterThanOrEqualTo(4L));
});
client.flush();
Awaitility.await().untilAsserted(() -> {
- assertEquals(mockRecord.failed, 0);
- assertEquals(mockRecord.acked, 5);
+ assertEquals(mockRecord.getFailed(), 0);
+ assertEquals(mockRecord.getAcked(), 5);
assertEquals(client.totalHits(index), 5);
});
@@ -309,10 +316,11 @@ public class ElasticSearchClientTests extends ElasticSearchTestBase {
log.info("elapsed = {}", elapsed);
assertTrue(elapsed > 29000); // bulkIndex was blocking while elasticsearch was down or busy
- Thread.sleep(1000L);
- assertEquals(mockRecord.acked, 15);
- assertEquals(mockRecord.failed, 0);
- assertEquals(client.records.size(), 0);
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(mockRecord.getAcked(), 15);
+ assertEquals(mockRecord.getFailed(), 0);
+ assertEquals(client.records.size(), 0);
+ });
chaosContainer.stop();
} finally {