diff --git a/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java b/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java index 73fff1bb7e269..dcc5707afba64 100644 --- a/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java +++ b/buildtools/src/main/java/org/apache/pulsar/tests/MockitoCleanupListener.java @@ -39,7 +39,8 @@ public class MockitoCleanupListener extends BetweenTestClassesListenerAdapter { protected void onBetweenTestClasses(Class endedTestClass, Class startedTestClass) { if (MOCKITO_CLEANUP_ENABLED) { if (MockitoThreadLocalStateCleaner.INSTANCE.isEnabled()) { - LOG.info("Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER thread local state."); + LOG.info("Cleaning up Mockito's ThreadSafeMockingProgress.MOCKING_PROGRESS_PROVIDER " + + "thread local state."); MockitoThreadLocalStateCleaner.INSTANCE.cleanup(); } cleanupMockitoInline(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java index da3043e7458bb..d1bc95c768a85 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/PositionAckSetUtil.java @@ -60,6 +60,13 @@ public static long[] andAckSet(long[] firstAckSet, long[] secondAckSet) { return ackSet; } + public static boolean isAckSetEmpty(long[] ackSet) { + BitSetRecyclable bitSet = BitSetRecyclable.create().resetWords(ackSet); + boolean isEmpty = bitSet.isEmpty(); + bitSet.recycle(); + return isEmpty; + } + //This method is compare two position which position is bigger than another one. //When the ledgerId and entryId in this position is same to another one and two position all have ack set, it will //compare the ack set next bit index is bigger than another one. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index 1be743a7ee93b..51525303ac0af 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -5068,7 +5068,8 @@ private void internalGetReplicatedSubscriptionStatusForNonPartitionedTopic(Async // Redirect the request to the appropriate broker if this broker is not the owner of the topic validateTopicOwnership(topicName, authoritative); - getReplicatedSubscriptionStatusFromLocalBroker(topicName, subName).get(); + Map res = getReplicatedSubscriptionStatusFromLocalBroker(topicName, subName).get(); + asyncResponse.resume(res); } catch (Exception e) { log.error("[{}] Failed to get replicated subscription status on {} {}", clientAppId(), topicName, subName, e); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java index 1beb8ec741524..d81f6949f43f7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java @@ -64,8 +64,8 @@ import org.apache.pulsar.broker.resources.ClusterResources; import org.apache.pulsar.broker.stats.prometheus.metrics.Summary; import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.client.util.ExecutorProvider; +import org.apache.pulsar.common.naming.NamespaceBundle; import org.apache.pulsar.common.naming.NamespaceBundleFactory; import org.apache.pulsar.common.naming.NamespaceName; import org.apache.pulsar.common.naming.ServiceUnitId; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java index 6b9ddcc1162f3..0b5108eeab861 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java @@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service; import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.andAckSet; +import static org.apache.bookkeeper.mledger.util.PositionAckSetUtil.isAckSetEmpty; import com.google.common.collect.ImmutableList; import io.netty.buffer.ByteBuf; import java.util.ArrayList; @@ -246,6 +247,18 @@ public int filterEntriesForConsumer(@Nullable EntryWrapper[] entryWrapper, int e // if actSet is null, use pendingAck ackSet ackSet = positionInPendingAck.getAckSet(); } + // if the result of pendingAckSet(in pendingAckHandle) AND the ackSet(in cursor) is empty + // filter this entry + if (isAckSetEmpty(ackSet)) { + entries.set(i, null); + entry.release(); + continue; + } + } else { + // filter non-batch message in pendingAck state + entries.set(i, null); + entry.release(); + continue; } } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java index 68e5b7da8243b..976ba5d328e75 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java @@ -21,10 +21,10 @@ import static com.google.common.base.Preconditions.checkArgument; import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.commons.lang3.StringUtils.isBlank; +import com.fasterxml.jackson.databind.ObjectMapper; import com.github.benmanes.caffeine.cache.CacheLoader; import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; -import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.BoundType; import com.google.common.collect.Lists; import com.google.common.collect.Range; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java index c0a468816281d..5de86180716c5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceOwnershipListenerTests.java @@ -19,6 +19,7 @@ package org.apache.pulsar.broker.namespace; import com.google.common.collect.Sets; +import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.broker.service.BrokerTestBase; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -30,11 +31,9 @@ import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; - import java.util.List; import java.util.UUID; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; import static org.testng.Assert.assertTrue; @@ -55,7 +54,7 @@ protected void cleanup() throws Exception { } @Test - public void testNamespaceBundleOwnershipListener() throws PulsarAdminException, InterruptedException, PulsarClientException { + public void testNamespaceBundleOwnershipListener() throws Exception { final CountDownLatch countDownLatch = new CountDownLatch(2); final AtomicBoolean onLoad = new AtomicBoolean(false); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index b0a4b28bbdc84..a51ad6998ed11 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -171,6 +171,84 @@ private void testIndividualAckAbortFilterAckSetInPendingAckState() throws Except assertNull(consumer.receive(2, TimeUnit.SECONDS)); } + + @Test(dataProvider="enableBatch") + private void testFilterMsgsInPendingAckStateWhenConsumerDisconnect(boolean enableBatch) throws Exception { + final String topicName = NAMESPACE1 + "/testFilterMsgsInPendingAckStateWhenConsumerDisconnect-" + enableBatch; + final int count = 10; + + @Cleanup + Producer producer = null; + if (enableBatch) { + producer = pulsarClient + .newProducer(Schema.INT32) + .topic(topicName) + .enableBatching(true) + .batchingMaxPublishDelay(1, TimeUnit.HOURS) + .batchingMaxMessages(count).create(); + } else { + producer = pulsarClient + .newProducer(Schema.INT32) + .topic(topicName) + .enableBatching(false).create(); + } + + @Cleanup + Consumer consumer = pulsarClient + .newConsumer(Schema.INT32) + .topic(topicName) + .isAckReceiptEnabled(true) + .subscriptionName("test") + .subscriptionType(SubscriptionType.Shared) + .enableBatchIndexAcknowledgment(true) + .subscribe(); + + for (int i = 0; i < count; i++) { + producer.sendAsync(i); + } + + Transaction txn1 = getTxn(); + + Transaction txn2 = getTxn(); + + + // txn1 ack half of messages and don't end the txn1 + for (int i = 0; i < count / 2; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn1).get(); + } + + // txn2 ack the rest half of messages and commit tnx2 + for (int i = count / 2; i < count; i++) { + consumer.acknowledgeAsync(consumer.receive().getMessageId(), txn2).get(); + } + // commit txn2 + txn2.commit().get(); + + // close and re-create consumer + consumer.close(); + consumer = pulsarClient + .newConsumer(Schema.INT32) + .topic(topicName) + .isAckReceiptEnabled(true) + .subscriptionName("test") + .subscriptionType(SubscriptionType.Shared) + .enableBatchIndexAcknowledgment(true) + .subscribe(); + + Message message = consumer.receive(3, TimeUnit.SECONDS); + Assert.assertNull(message); + + // abort txn1 + txn1.abort().get(); + // after txn1 aborted, consumer will receive messages txn1 contains + int receiveCounter = 0; + while((message = consumer.receive(3, TimeUnit.SECONDS)) != null) { + Assert.assertEquals(message.getValue().intValue(), receiveCounter); + receiveCounter ++; + } + Assert.assertEquals(receiveCounter, count / 2); + } + @Test(dataProvider="enableBatch") private void produceCommitTest(boolean enableBatch) throws Exception { @Cleanup diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java index d278a18769057..06b9c0e555b3b 100644 --- a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java +++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/PulsarAdminBuilderImplTest.java @@ -34,8 +34,6 @@ import org.apache.pulsar.client.api.EncodedAuthenticationParameterSupport; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.impl.auth.AuthenticationDisabled; -import org.apache.pulsar.client.impl.conf.ClientConfigurationData; -import org.testng.Assert; import org.testng.annotations.Test; /** diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java index 0f42f35608a76..b1a94c4916224 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/BitSetRecyclableRecyclableTest.java @@ -45,4 +45,22 @@ public void testResetWords() { Assert.assertTrue(bitset1.get(128)); Assert.assertFalse(bitset1.get(256)); } + + @Test + public void testBitSetEmpty() { + BitSetRecyclable bitSet = BitSetRecyclable.create(); + bitSet.set(0, 5); + bitSet.clear(1); + bitSet.clear(2); + bitSet.clear(3); + long[] array = bitSet.toLongArray(); + Assert.assertFalse(bitSet.isEmpty()); + Assert.assertFalse(BitSetRecyclable.create().resetWords(array).isEmpty()); + bitSet.clear(0); + bitSet.clear(4); + Assert.assertTrue(bitSet.isEmpty()); + long[] array1 = bitSet.toLongArray(); + Assert.assertTrue(BitSetRecyclable.create().resetWords(array1).isEmpty()); + bitSet.recycle(); + } } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java index 38d85335bcd52..7b73780caac8f 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java @@ -18,7 +18,6 @@ */ package org.apache.pulsar.functions.instance; -import io.netty.util.concurrent.DefaultThreadFactory; import lombok.Getter; import java.util.concurrent.Executors; diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java index d15d2ba90c557..5dbb61456e4d2 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/ClusterServiceCoordinator.java @@ -31,9 +31,6 @@ import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.util.ExecutorProvider; - -import java.util.HashMap; -import java.util.Map; import java.util.function.Supplier; @Slf4j diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java deleted file mode 100644 index e420d7a5cef99..0000000000000 --- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/client/elastic/ElasticSearchJavaRestClient.java +++ /dev/null @@ -1,224 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.io.elasticsearch.client.elastic; - -import co.elastic.clients.elasticsearch.ElasticsearchClient; -import co.elastic.clients.elasticsearch._types.ElasticsearchException; -import co.elastic.clients.elasticsearch._types.Result; -import co.elastic.clients.elasticsearch.core.DeleteRequest; -import co.elastic.clients.elasticsearch.core.DeleteResponse; -import co.elastic.clients.elasticsearch.core.IndexRequest; -import co.elastic.clients.elasticsearch.core.IndexResponse; -import co.elastic.clients.elasticsearch.core.SearchRequest; -import co.elastic.clients.elasticsearch.core.SearchResponse; -import co.elastic.clients.elasticsearch.indices.CreateIndexRequest; -import co.elastic.clients.elasticsearch.indices.CreateIndexResponse; -import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest; -import co.elastic.clients.elasticsearch.indices.ExistsRequest; -import co.elastic.clients.elasticsearch.indices.IndexSettings; -import co.elastic.clients.elasticsearch.indices.RefreshRequest; -import co.elastic.clients.json.jackson.JacksonJsonpMapper; -import co.elastic.clients.transport.ElasticsearchTransport; -import co.elastic.clients.transport.rest_client.RestClientTransport; -import com.fasterxml.jackson.annotation.JsonInclude; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.SerializationFeature; -import com.google.common.annotations.VisibleForTesting; -import java.io.IOException; -import java.util.Map; -import java.util.Objects; -import lombok.extern.slf4j.Slf4j; -import org.apache.http.HttpHost; -import org.apache.pulsar.io.elasticsearch.ElasticSearchConfig; -import org.apache.pulsar.io.elasticsearch.client.BulkProcessor; -import org.apache.pulsar.io.elasticsearch.client.RestClient; -import org.elasticsearch.client.Node; -import org.elasticsearch.client.RestClientBuilder; -import org.opensearch.action.bulk.BulkProcessor; - -@Slf4j -public class ElasticSearchJavaRestClient extends RestClient { - - private final ElasticsearchClient client; - private final ObjectMapper objectMapper = new ObjectMapper() - .configure(SerializationFeature.INDENT_OUTPUT, false) - .setSerializationInclusion(JsonInclude.Include.ALWAYS); - private BulkProcessor bulkProcessor; - private ElasticsearchTransport transport; - - @VisibleForTesting - public void setBulkProcessor(BulkProcessor bulkProcessor) { - this.bulkProcessor = bulkProcessor; - } - - @VisibleForTesting - public void setTransport(ElasticsearchTransport transport) { - this.transport = transport; - } - - public ElasticSearchJavaRestClient(ElasticSearchConfig elasticSearchConfig, - BulkProcessor.Listener bulkProcessorListener) { - super(elasticSearchConfig, bulkProcessorListener); - - log.info("ElasticSearch URL {}", config.getElasticSearchUrl()); - final HttpHost[] httpHosts = getHttpHosts(); - - RestClientBuilder builder = org.elasticsearch.client.RestClient.builder(httpHosts) - .setRequestConfigCallback(builder1 -> builder1 - .setContentCompressionEnabled(config.isCompressionEnabled()) - .setConnectionRequestTimeout(config.getConnectionRequestTimeoutInMs()) - .setConnectTimeout(config.getConnectTimeoutInMs()) - .setSocketTimeout(config.getSocketTimeoutInMs())) - .setHttpClientConfigCallback(this.configCallback) - .setFailureListener(new org.elasticsearch.client.RestClient.FailureListener() { - public void onFailure(Node node) { - log.warn("Node host={} failed", node.getHost()); - } - }); - transport = new RestClientTransport(builder.build(), new JacksonJsonpMapper(objectMapper)); - client = new ElasticsearchClient(transport); - if (elasticSearchConfig.isBulkEnabled()) { - bulkProcessor = new ElasticBulkProcessor(elasticSearchConfig, client, bulkProcessorListener); - } else { - bulkProcessor = null; - } - } - - @Override - public boolean indexExists(String index) throws IOException { - final ExistsRequest request = new ExistsRequest.Builder() - .index(index) - .build(); - return client.indices().exists(request).value(); - } - - @Override - public boolean createIndex(String index) throws IOException { - final CreateIndexRequest createIndexRequest = new CreateIndexRequest.Builder() - .index(index) - .settings(new IndexSettings.Builder() - .numberOfShards(config.getIndexNumberOfShards() + "") - .numberOfReplicas(config.getIndexNumberOfReplicas() + "") - .build() - ) - .build(); - try { - final CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest); - if ((createIndexResponse.acknowledged()) - && createIndexResponse.shardsAcknowledged()) { - return true; - } - throw new IOException("Unable to create index, acknowledged: " + createIndexResponse.acknowledged() - + " shardsAcknowledged: " + createIndexResponse.shardsAcknowledged()); - } catch (ElasticsearchException ex) { - final String errorType = Objects.requireNonNull(ex.response().error().type()); - if (errorType.contains("resource_already_exists_exception")) { - return false; - } - throw ex; - } - } - - @Override - public boolean deleteIndex(String index) throws IOException { - return client.indices().delete(new DeleteIndexRequest.Builder().index(index).build()).acknowledged(); - } - - @Override - public boolean deleteDocument(String index, String documentId) throws IOException { - final DeleteRequest req = new - DeleteRequest.Builder() - .index(config.getIndexName()) - .id(documentId) - .build(); - - DeleteResponse deleteResponse = client.delete(req); - return deleteResponse.result().equals(Result.Deleted) || deleteResponse.result().equals(Result.NotFound); - } - - @Override - public boolean indexDocument(String index, String documentId, String documentSource) throws IOException { - final Map mapped = objectMapper.readValue(documentSource, Map.class); - final IndexRequest indexRequest = new IndexRequest.Builder<>() - .index(config.getIndexName()) - .document(mapped) - .id(documentId) - .build(); - final IndexResponse indexResponse = client.index(indexRequest); - - return indexResponse.result().equals(Result.Created) || indexResponse.result().equals(Result.Updated); - } - - public SearchResponse search(String indexName) throws IOException { - return search(indexName, "*:*"); - } - - @VisibleForTesting - public SearchResponse search(String indexName, String query) throws IOException { - final RefreshRequest refreshRequest = new RefreshRequest.Builder().index(indexName).build(); - client.indices().refresh(refreshRequest); - - query = query.replace("/", "\\/"); - return client.search(new SearchRequest.Builder().index(indexName) - .q(query) - .build(), Map.class); - } - - @Override - public long totalHits(String indexName) throws IOException { - return totalHits(indexName, "*:*"); - } - - @Override - public long totalHits(String indexName, String query) throws IOException { - final SearchResponse searchResponse = search(indexName, query); - return searchResponse.hits().total().value(); - } - - @Override - public BulkProcessor getBulkProcessor() { - if (bulkProcessor == null) { - throw new IllegalStateException("bulkProcessor not enabled"); - } - return bulkProcessor; - } - - @Override - public void closeClient() { - if (bulkProcessor != null) { - bulkProcessor.close(); - } - // client doesn't need to be closed, only the transport instance - try { - transport.close(); - } catch (IOException e) { - log.warn("error while closing the client", e); - } - } - - @VisibleForTesting - public ElasticsearchClient getClient() { - return client; - } - - @VisibleForTesting - public ElasticsearchTransport getTransport() { - return transport; - } -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java deleted file mode 100644 index 65b38c677bfc5..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch7SinkTester.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.tests.integration.io.sinks; - -import java.util.Optional; -import org.apache.pulsar.tests.integration.topologies.PulsarCluster; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -public class ElasticSearch7SinkTester extends ElasticSearchSinkTester { - - public static final String ELASTICSEARCH_7 = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE_V7")) - .orElse("docker.elastic.co/elasticsearch/elasticsearch:7.17.7"); - - - public ElasticSearch7SinkTester(boolean schemaEnable) { - super(schemaEnable); - } - - @Override - protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { - return new ElasticsearchContainer(ELASTICSEARCH_7) - .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m"); - } - -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java deleted file mode 100644 index bb52c4ff03fea..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/ElasticSearch8SinkTester.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.tests.integration.io.sinks; - -import java.util.Optional; -import org.apache.pulsar.tests.integration.topologies.PulsarCluster; -import org.testcontainers.elasticsearch.ElasticsearchContainer; - -public class ElasticSearch8SinkTester extends ElasticSearchSinkTester { - - public static final String ELASTICSEARCH_8 = Optional.ofNullable(System.getenv("ELASTICSEARCH_IMAGE_V8")) - .orElse("docker.elastic.co/elasticsearch/elasticsearch:8.5.1"); - - - public ElasticSearch8SinkTester(boolean schemaEnable) { - super(schemaEnable); - } - - @Override - protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { - return new ElasticsearchContainer(ELASTICSEARCH_8) - .withEnv("ES_JAVA_OPTS", "-Xms128m -Xmx256m") - .withEnv("xpack.security.enabled", "false") - .withEnv("xpack.security.http.ssl.enabled", "false"); - } - -} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java deleted file mode 100644 index 1e10cc4189c1a..0000000000000 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/OpenSearchSinkTester.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.pulsar.tests.integration.io.sinks; - -import java.util.Optional; -import org.apache.http.HttpHost; -import org.apache.pulsar.tests.integration.topologies.PulsarCluster; -import org.awaitility.Awaitility; -import org.opensearch.action.search.SearchRequest; -import org.opensearch.action.search.SearchResponse; -import org.opensearch.client.RequestOptions; -import org.opensearch.client.RestClient; -import org.opensearch.client.RestClientBuilder; -import org.opensearch.client.RestHighLevelClient; -import org.testcontainers.elasticsearch.ElasticsearchContainer; -import org.testcontainers.utility.DockerImageName; - -import java.util.Map; - -import static org.testng.Assert.assertTrue; - -public class OpenSearchSinkTester extends ElasticSearchSinkTester { - - public static final String OPENSEARCH = Optional.ofNullable(System.getenv("OPENSEARCH_IMAGE")) - .orElse("opensearchproject/opensearch:1.2.4"); - - private RestHighLevelClient elasticClient; - - - public OpenSearchSinkTester(boolean schemaEnable) { - super(schemaEnable); - } - - @Override - protected ElasticsearchContainer createSinkService(PulsarCluster cluster) { - DockerImageName dockerImageName = DockerImageName.parse(OPENSEARCH) - .asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch"); - return new ElasticsearchContainer(dockerImageName) - .withEnv("OPENSEARCH_JAVA_OPTS", "-Xms128m -Xmx256m") - .withEnv("bootstrap.memory_lock", "true") - .withEnv("plugins.security.disabled", "true"); - } - - @Override - public void prepareSink() throws Exception { - RestClientBuilder builder = RestClient.builder( - new HttpHost( - "localhost", - serviceContainer.getMappedPort(9200), - "http")); - elasticClient = new RestHighLevelClient(builder); - } - - @Override - public void validateSinkResult(Map kvs) { - org.opensearch.action.search.SearchRequest searchRequest = new SearchRequest("test-index"); - - Awaitility.await().untilAsserted(() -> { - SearchResponse searchResult = elasticClient.search(searchRequest, RequestOptions.DEFAULT); - assertTrue(searchResult.getHits().getTotalHits().value > 0, searchResult.toString()); - }); - } - - @Override - public void close() throws Exception { - super.close(); - if (elasticClient != null) { - elasticClient.close(); - elasticClient = null; - } - } -}