You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/10/26 03:04:34 UTC
[pulsar] branch branch-2.8 updated (1a869ef -> 8bea3cd)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 1a869ef Add debug log for WebSocket. (#12458)
new eb91b36 Future completed twice in the method of impl.MLPendingAckStore#closeAsync (#12362)
new 232d61b optimize SecurityUtility code flow (#12431)
new 9aadc7e broker resource group test optimize fail msg (#12438)
new e7c3bfb [Pulsar SQL] Pulsar SQL support query big entry data (#12448)
new 4884fb7 fix a typo in UnAckedMessageTracker (#12467)
new 83d941f Optimize the code: remove extra spaces (#12470)
new f07a87d fix the race of delete subscription and delete topic (#12240)
new 39ee36c #12429 only fixed the compactor skips data issue, but the normal reader/consumer (#12464)
new 8bea3cd Fix the retry topic's `REAL_TOPIC` & `ORIGIN_MESSAGE_ID` property should not be modified once it has been written. (#12451)
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
conf/presto/catalog/pulsar.properties | 3 +-
.../broker/service/persistent/PersistentTopic.java | 5 +
.../pendingack/impl/MLPendingAckStore.java | 23 +++--
.../pulsar/compaction/CompactedTopicImpl.java | 4 +-
.../ResourceGroupUsageAggregationTest.java | 15 ++-
.../ResourceUsageTransportManagerTest.java | 1 -
.../apache/pulsar/client/api/RetryTopicTest.java | 105 ++++++++++++++++++++-
.../pulsar/compaction/CompactedTopicTest.java | 15 +++
.../apache/pulsar/client/impl/ConsumerBase.java | 2 +-
.../apache/pulsar/client/impl/ConsumerImpl.java | 4 +-
.../pulsar/client/impl/NegativeAcksTracker.java | 4 +-
.../pulsar/client/impl/UnAckedMessageTracker.java | 11 +--
.../apache/pulsar/common/util/SecurityUtility.java | 12 +--
.../pulsar/sql/presto/PulsarConnectorCache.java | 4 +-
site2/docs/sql-deployment-configurations.md | 3 +
.../tests/integration/containers/BKContainer.java | 1 +
.../tests/integration/presto/TestBasicPresto.java | 66 ++++++++-----
.../presto/TestPrestoQueryTieredStorage.java | 9 +-
.../integration/presto/TestPulsarSQLBase.java | 18 ++--
.../integration/suites/PulsarSQLTestSuite.java | 52 ++++++++++
.../integration/topologies/PulsarCluster.java | 6 +-
.../integration/topologies/PulsarClusterSpec.java | 4 +
22 files changed, 283 insertions(+), 84 deletions(-)
[pulsar] 09/09: Fix the retry topic's `REAL_TOPIC` &
`ORIGIN_MESSAGE_ID` property should not be modified once it has been
written. (#12451)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8bea3cd5ec18ef6fb9d89dfbf261f486b59c7095
Author: YANGLiiN <ie...@qq.com>
AuthorDate: Mon Oct 25 12:17:57 2021 +0800
Fix the retry topic's `REAL_TOPIC` & `ORIGIN_MESSAGE_ID` property should not be modified once it has been written. (#12451)
### Motivation
when reconsumer the message with the configuration maxRedeliveryCount > 1 ,eg deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(3).build()):
then when consumer the same message by the third time (RECONSUMETIMES=2) the REAL_TOPIC and the ORIGIN_MESSAGE_ID changed as the second message.
but, this should not changed once the REAL_TOPIC ORIGIN_MESSAGE_ID property was written .
```
2021-10-21T15:03:12,771+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 4:0:-1:0 {}
2021-10-21T15:03:12,909+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 3:1:-1 {REAL_TOPIC=persistent://my-property/my-ns/retry-topic, ORIGIN_MESSAGE_IDY_TIME=4:1:-1:0, DELAY_TIME=1000, RECONSUMETIMES=1}
2021-10-21T15:03:12,965+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 3:10:-1 {REAL_TOPIC=persistent://my-property/my-ns/retry-topic-my-subscription-RETRY, ORIGIN_MESSAGE_IDY_TIME=3:0:-1, DELAY_TIME=1000, RECONSUMETIMES=2}
2021-10-21T15:03:13,026+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 3:20:-1 {REAL_TOPIC=persistent://my-property/my-ns/retry-topic-my-subscription-RETRY, ORIGIN_MESSAGE_IDY_TIME=3:10:-1, DELAY_TIME=1000, RECONSUMETIMES=3}
```
Expected Results (REAL_TOPIC ORIGIN_MESSAGE_ID property equals the frist origin message)
```
2021-10-21T15:27:01,390+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 4:0:-1:0 {}
2021-10-21T15:27:01,479+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 3:0:-1 {REAL_TOPIC=persistent://my-property/my-ns/retry-topic, ORIGIN_MESSAGE_IDY_TIME=4:0:-1:0, DELAY_TIME=1000, RECONSUMETIMES=1}
2021-10-21T15:27:01,547+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 3:10:-1 {REAL_TOPIC=persistent://my-property/my-ns/retry-topic, ORIGIN_MESSAGE_IDY_TIME=4:0:-1:0, DELAY_TIME=1000, RECONSUMETIMES=2}
2021-10-21T15:27:01,603+0800 [main] ERROR org.apache.pulsar.client.api.RetryTopicTest - consumer received message : 3:20:-1 {REAL_TOPIC=persistent://my-property/my-ns/retry-topic, ORIGIN_MESSAGE_IDY_TIME=4:0:-1:0, DELAY_TIME=1000, RECONSUMETIMES=3}
```
### Modifications
https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L652-L653
propertiesMap.put() --> propertiesMap.putIfAbsent()
add one testcase to verify the REAL_TOPIC ORIGIN_MESSAGE_ID property
(cherry picked from commit 85178830c9bf7462729f6f5d8d9285dee5d2d93d)
---
.../apache/pulsar/client/api/RetryTopicTest.java | 105 ++++++++++++++++++++-
.../apache/pulsar/client/impl/ConsumerImpl.java | 4 +-
2 files changed, 102 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
index fc84a62..48d20c6 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/RetryTopicTest.java
@@ -18,18 +18,19 @@
*/
package org.apache.pulsar.client.api;
-import lombok.Cleanup;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import com.google.common.collect.Sets;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.client.util.RetryMessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
-import java.util.concurrent.TimeUnit;
-
-import static org.testng.Assert.assertNull;
-
@Test(groups = "broker-api")
public class RetryTopicTest extends ProducerConsumerBase {
@@ -119,6 +120,100 @@ public class RetryTopicTest extends ProducerConsumerBase {
checkConsumer.close();
}
+ @Test
+ public void testRetryTopicProperties() throws Exception {
+ final String topic = "persistent://my-property/my-ns/retry-topic";
+
+ final int maxRedeliveryCount = 3;
+
+ final int sendMessages = 10;
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic)
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Shared)
+ .enableRetry(true)
+ .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).build())
+ .receiverQueueSize(100)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ @Cleanup
+ PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0);
+ Consumer<byte[]> deadLetterConsumer = newPulsarClient.newConsumer(Schema.BYTES)
+ .topic("persistent://my-property/my-ns/retry-topic-my-subscription-DLQ")
+ .subscriptionName("my-subscription")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .create();
+
+ Set<String> originMessageIds = Sets.newHashSet();
+ for (int i = 0; i < sendMessages; i++) {
+ MessageId msgId = producer.send(String.format("Hello Pulsar [%d]", i).getBytes());
+ originMessageIds.add(msgId.toString());
+ }
+
+ producer.close();
+
+ int totalReceived = 0;
+ Set<String> retryMessageIds = Sets.newHashSet();
+ do {
+ Message<byte[]> message = consumer.receive();
+ log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
+ // retry message
+ if (message.hasProperty(RetryMessageUtil.SYSTEM_PROPERTY_RECONSUMETIMES)) {
+ // check the REAL_TOPIC property
+ assertEquals(message.getProperty(RetryMessageUtil.SYSTEM_PROPERTY_REAL_TOPIC), topic);
+ retryMessageIds.add(message.getProperty(RetryMessageUtil.SYSTEM_PROPERTY_ORIGIN_MESSAGE_ID));
+ }
+ consumer.reconsumeLater(message, 1, TimeUnit.SECONDS);
+ totalReceived++;
+ } while (totalReceived < sendMessages * (maxRedeliveryCount + 1));
+
+ // check the REAL_TOPIC property
+ assertEquals(retryMessageIds, originMessageIds);
+
+ int totalInDeadLetter = 0;
+ Set<String> deadLetterMessageIds = Sets.newHashSet();
+ do {
+ Message message = deadLetterConsumer.receive();
+ log.info("dead letter consumer received message : {} {}", message.getMessageId(),
+ new String(message.getData()));
+ // dead letter message
+ if (message.hasProperty(RetryMessageUtil.SYSTEM_PROPERTY_RECONSUMETIMES)) {
+ // check the REAL_TOPIC property
+ assertEquals(message.getProperty(RetryMessageUtil.SYSTEM_PROPERTY_REAL_TOPIC), topic);
+ deadLetterMessageIds.add(message.getProperty(RetryMessageUtil.SYSTEM_PROPERTY_ORIGIN_MESSAGE_ID));
+ }
+ deadLetterConsumer.acknowledge(message);
+ totalInDeadLetter++;
+ } while (totalInDeadLetter < sendMessages);
+
+ assertEquals(deadLetterMessageIds, originMessageIds);
+
+ deadLetterConsumer.close();
+ consumer.close();
+
+ Consumer<byte[]> checkConsumer = this.pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic)
+ .subscriptionName("my-subscription")
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ Message<byte[]> checkMessage = checkConsumer.receive(3, TimeUnit.SECONDS);
+ if (checkMessage != null) {
+ log.info("check consumer received message : {} {}", checkMessage.getMessageId(),
+ new String(checkMessage.getData()));
+ }
+ assertNull(checkMessage);
+
+ checkConsumer.close();
+ }
+
//Issue 9327: do compatibility check in case of the default retry and dead letter topic name changed
@Test
public void testRetryTopicNameForCompatibility () throws Exception {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 871f840..3835950 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -641,8 +641,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (message.getProperties() != null) {
propertiesMap.putAll(message.getProperties());
}
- propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_REAL_TOPIC, originTopicNameStr);
- propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_ORIGIN_MESSAGE_ID, originMessageIdStr);
+ propertiesMap.putIfAbsent(RetryMessageUtil.SYSTEM_PROPERTY_REAL_TOPIC, originTopicNameStr);
+ propertiesMap.putIfAbsent(RetryMessageUtil.SYSTEM_PROPERTY_ORIGIN_MESSAGE_ID, originMessageIdStr);
return propertiesMap;
}
[pulsar] 08/09: #12429 only fixed the compactor skips data issue,
but the normal reader/consumer (#12464)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 39ee36cf3e33002a6e8bc5ba7eb5060e239da7c2
Author: lipenghui <pe...@apache.org>
AuthorDate: Sun Oct 24 23:11:15 2021 +0800
#12429 only fixed the compactor skips data issue, but the normal reader/consumer (#12464)
also skips data while enabled read compacted data and read from the earliest position.
(cherry picked from commit dd90657c33aa71ca86c9988513d92bf79aa68335)
---
.../org/apache/pulsar/compaction/CompactedTopicImpl.java | 4 +---
.../org/apache/pulsar/compaction/CompactedTopicTest.java | 15 +++++++++++++++
2 files changed, 16 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index 4375188..4bc1664 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -18,7 +18,6 @@
*/
package org.apache.pulsar.compaction;
-import static org.apache.pulsar.compaction.Compactor.COMPACTION_SUBSCRIPTION;
import com.github.benmanes.caffeine.cache.AsyncLoadingCache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.ComparisonChain;
@@ -128,8 +127,7 @@ public class CompactedTopicImpl implements CompactedTopic {
// need to force seek the read position to ensure the compactor can read
// the complete last snapshot because of the compactor will read the data
// before the compaction cursor mark delete position
- cursor.seek(lastEntry.getPosition().getNext(),
- cursor.getName().equals(COMPACTION_SUBSCRIPTION));
+ cursor.seek(lastEntry.getPosition().getNext(), true);
callback.readEntriesComplete(entries, consumer);
});
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index adb6f46..2dd6f8a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -481,5 +481,20 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
PersistentTopicInternalStats stats = admin.topics().getInternalStats(topic);
Assert.assertEquals(stats.compactedLedger.entries, keys + 1);
});
+
+ // Make sure the reader can get all data from the compacted ledger and original ledger.
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .startMessageId(MessageId.earliest)
+ .readCompacted(true)
+ .create();
+ int received = 0;
+ while (reader.hasMessageAvailable()) {
+ reader.readNext();
+ received++;
+ }
+ Assert.assertEquals(received, keys + 1);
+ reader.close();
+ producer.close();
}
}
[pulsar] 03/09: broker resource group test optimize fail msg
(#12438)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9aadc7ea4b676296a3a2daf04a0cf5983208b6ae
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Sun Oct 24 12:41:55 2021 +0800
broker resource group test optimize fail msg (#12438)
(cherry picked from commit 86f40c1dc127aa9af928c544ff4062666c621ef4)
---
.../resourcegroup/ResourceGroupUsageAggregationTest.java | 15 +++++++--------
.../resourcegroup/ResourceUsageTransportManagerTest.java | 1 -
2 files changed, 7 insertions(+), 9 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupUsageAggregationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupUsageAggregationTest.java
index e2afb14..5b3db89 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupUsageAggregationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceGroupUsageAggregationTest.java
@@ -126,8 +126,8 @@ public class ResourceGroupUsageAggregationTest extends ProducerConsumerBase {
.subscriptionType(SubscriptionType.Shared)
.subscribe();
} catch (PulsarClientException p) {
- final String errMesg = String.format("Got exception while building consumer: ex={}", p.getMessage());
- Assert.assertTrue(false, errMesg);
+ final String errMsg = String.format("Got exception while building consumer: ex=%s", p.getMessage());
+ Assert.fail(errMsg);
}
final TopicName myTopic = TopicName.get(topicString);
@@ -142,17 +142,16 @@ public class ResourceGroupUsageAggregationTest extends ProducerConsumerBase {
int recvdNumBytes = 0;
int recvdNumMsgs = 0;
for (int ix = 0; ix < NumMessagesToSend; ix++) {
- MessageId prodMesgId = null;
byte[] mesg;
try {
- mesg = String.format("Hi, ix={}", ix).getBytes();
+ mesg = String.format("Hi, ix=%s", ix).getBytes();
producer.send(mesg);
sentNumBytes += mesg.length;
sentNumMsgs++;
this.verfyStats(topicString, rgConfig.getName(), sentNumBytes, sentNumMsgs, recvdNumBytes, recvdNumMsgs, true, false);
} catch (PulsarClientException p) {
- final String errMesg = String.format("Got exception while sending {}-th time: ex={}", ix, p.getMessage());
- Assert.assertTrue(false, errMesg);
+ final String errMsg = String.format("Got exception while sending %s-th time: ex=%s", ix, p.getMessage());
+ Assert.fail(errMsg);
}
}
producer.close();
@@ -165,9 +164,9 @@ public class ResourceGroupUsageAggregationTest extends ProducerConsumerBase {
message = consumer.receive();
recvdNumBytes += message.getValue().length;
} catch (PulsarClientException p) {
- final String errMesg = String.format("Got exception in while receiving {}-th mesg at consumer: ex={}",
+ final String errMesg = String.format("Got exception in while receiving %s-th mesg at consumer: ex=%s",
recvdNumMsgs, p.getMessage());
- Assert.assertTrue(false, errMesg);
+ Assert.fail(errMesg);
}
// log.info("consumer received message : {} {}", message.getMessageId(), new String(message.getData()));
recvdNumMsgs++;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTransportManagerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTransportManagerTest.java
index a6fa17f..588f9c9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTransportManagerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTransportManagerTest.java
@@ -24,7 +24,6 @@ import org.apache.pulsar.broker.service.resource.usage.ResourceUsage;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
[pulsar] 07/09: fix the race of delete subscription and delete
topic (#12240)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f07a87d0926f73a5f8cb62e4560b332f3718232f
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Mon Oct 25 08:41:05 2021 +0800
fix the race of delete subscription and delete topic (#12240)
(cherry picked from commit 9137e6b1e561cd1bfc4540e651aa33e06934acf8)
---
.../org/apache/pulsar/broker/service/persistent/PersistentTopic.java | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 30315fb..6c083e8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1020,6 +1020,11 @@ public class PersistentTopic extends AbstractTopic
log.debug("[{}][{}] Error deleting cursor for subscription",
topic, subscriptionName, exception);
}
+ if (exception instanceof ManagedLedgerException.ManagedLedgerNotFoundException) {
+ unsubscribeFuture.complete(null);
+ lastActive = System.nanoTime();
+ return;
+ }
unsubscribeFuture.completeExceptionally(new PersistenceException(exception));
}
}, null);
[pulsar] 01/09: Future completed twice in the method of
impl.MLPendingAckStore#closeAsync (#12362)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eb91b363eadd2077ed8b9a659b7e7781600728f3
Author: chenlin <15...@qq.com>
AuthorDate: Fri Oct 22 16:23:42 2021 +0800
Future completed twice in the method of impl.MLPendingAckStore#closeAsync (#12362)
(cherry picked from commit 10371ee870d26a079d7d744154e7c4f67ac09fae)
---
.../pendingack/impl/MLPendingAckStore.java | 23 ++++++++++++++++------
1 file changed, 17 insertions(+), 6 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
index b60f4ba..fb88878 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/pendingack/impl/MLPendingAckStore.java
@@ -120,12 +120,23 @@ public class MLPendingAckStore implements PendingAckStore {
cursor.asyncClose(new AsyncCallbacks.CloseCallback() {
@Override
public void closeComplete(Object ctx) {
- try {
- managedLedger.close();
- } catch (Exception e) {
- completableFuture.completeExceptionally(e);
- }
- completableFuture.complete(null);
+ managedLedger.asyncClose(new AsyncCallbacks.CloseCallback() {
+
+ @Override
+ public void closeComplete(Object ctx) {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}][{}] MLPendingAckStore closed successfully！", managedLedger.getName(), ctx);
+ }
+ completableFuture.complete(null);
+ }
+
+ @Override
+ public void closeFailed(ManagedLedgerException exception, Object ctx) {
+ log.error("[{}][{}] MLPendingAckStore closed failed,exception={}", managedLedger.getName(),
+ ctx, exception);
+ completableFuture.completeExceptionally(exception);
+ }
+ }, ctx);
}
@Override
[pulsar] 02/09: optimize SecurityUtility code flow (#12431)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 232d61b5e01807cfdaff943b93aeb61ef6d9fef7
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Sun Oct 24 12:24:06 2021 +0800
optimize SecurityUtility code flow (#12431)
(cherry picked from commit bc339bb6d72c3fc0dea200ecb1cc035943873076)
---
.../java/org/apache/pulsar/common/util/SecurityUtility.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
index 52a5793..fb6c835 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/SecurityUtility.java
@@ -422,12 +422,12 @@ public class SecurityUtility {
}
public static PrivateKey loadPrivateKeyFromPemFile(String keyFilePath) throws KeyManagementException {
- PrivateKey privateKey = null;
-
if (keyFilePath == null || keyFilePath.isEmpty()) {
- return privateKey;
+ return null;
}
+ PrivateKey privateKey;
+
try (FileInputStream input = new FileInputStream(keyFilePath)) {
privateKey = loadPrivateKeyFromPemStream(input);
} catch (IOException e) {
@@ -438,12 +438,12 @@ public class SecurityUtility {
}
public static PrivateKey loadPrivateKeyFromPemStream(InputStream inStream) throws KeyManagementException {
- PrivateKey privateKey = null;
-
if (inStream == null) {
- return privateKey;
+ return null;
}
+ PrivateKey privateKey;
+
try (BufferedReader reader = new BufferedReader(new InputStreamReader(inStream, StandardCharsets.UTF_8))) {
if (inStream.markSupported()) {
inStream.reset();
[pulsar] 04/09: [Pulsar SQL] Pulsar SQL support query big entry
data (#12448)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e7c3bfb8bb8b3a97d4da65fa6093356151c5ec7b
Author: ran <ga...@126.com>
AuthorDate: Tue Oct 26 08:24:33 2021 +0800
[Pulsar SQL] Pulsar SQL support query big entry data (#12448)
(cherry picked from commit a8f0788915393e438e5c38aa9497080fea19d733)
---
conf/presto/catalog/pulsar.properties | 3 +-
.../pulsar/sql/presto/PulsarConnectorCache.java | 4 +-
site2/docs/sql-deployment-configurations.md | 3 +
.../tests/integration/containers/BKContainer.java | 1 +
.../tests/integration/presto/TestBasicPresto.java | 66 ++++++++++++++--------
.../presto/TestPrestoQueryTieredStorage.java | 9 +--
.../integration/presto/TestPulsarSQLBase.java | 18 +++---
.../integration/suites/PulsarSQLTestSuite.java | 52 +++++++++++++++++
.../integration/topologies/PulsarCluster.java | 6 +-
.../integration/topologies/PulsarClusterSpec.java | 4 ++
10 files changed, 122 insertions(+), 44 deletions(-)
diff --git a/conf/presto/catalog/pulsar.properties b/conf/presto/catalog/pulsar.properties
index 5248678..0348fcd 100644
--- a/conf/presto/catalog/pulsar.properties
+++ b/conf/presto/catalog/pulsar.properties
@@ -42,7 +42,8 @@ pulsar.max-split-queue-cache-size=-1
# to prevent erroneous rewriting
pulsar.namespace-delimiter-rewrite-enable=false
pulsar.rewrite-namespace-delimiter=/
-
+# max size of one batch message (default value is 5MB)
+# pulsar.max-message-size=5242880
####### TIERED STORAGE OFFLOADER CONFIGS #######
diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
index 4577099..e41524c 100644
--- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
+++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorCache.java
@@ -42,6 +42,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.PulsarVersion;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
+import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.metadata.api.MetadataStore;
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.MetadataStoreFactory;
@@ -110,7 +111,8 @@ public class PulsarConnectorCache {
.setReadEntryTimeout(60)
.setThrottleValue(pulsarConnectorConfig.getBookkeeperThrottleValue())
.setNumIOThreads(pulsarConnectorConfig.getBookkeeperNumIOThreads())
- .setNumWorkerThreads(pulsarConnectorConfig.getBookkeeperNumWorkerThreads());
+ .setNumWorkerThreads(pulsarConnectorConfig.getBookkeeperNumWorkerThreads())
+ .setNettyMaxFrameSizeBytes(pulsarConnectorConfig.getMaxMessageSize() + Commands.MESSAGE_SIZE_FRAME_PADDING);
ManagedLedgerFactoryConfig managedLedgerFactoryConfig = new ManagedLedgerFactoryConfig();
managedLedgerFactoryConfig.setMaxCacheSize(pulsarConnectorConfig.getManagedLedgerCacheSizeMB());
diff --git a/site2/docs/sql-deployment-configurations.md b/site2/docs/sql-deployment-configurations.md
index 1fe0353..e5c402e 100644
--- a/site2/docs/sql-deployment-configurations.md
+++ b/site2/docs/sql-deployment-configurations.md
@@ -24,6 +24,9 @@ pulsar.entry-read-batch-size=100
# default number of splits to use per query
pulsar.target-num-splits=4
+
+# max size of one batch message (default value is 5MB)
+pulsar.max-message-size=5242880
```
You can connect Presto to a Pulsar cluster with multiple hosts. To configure multiple hosts for brokers, add multiple URLs to `pulsar.web-service-url`. To configure multiple hosts for ZooKeeper, add multiple URIs to `pulsar.zookeeper-uri`. The following is an example.
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BKContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BKContainer.java
index 36f17cd..b294cac4 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BKContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/BKContainer.java
@@ -28,5 +28,6 @@ public class BKContainer extends PulsarContainer<BKContainer> {
public BKContainer(String clusterName, String hostName) {
super(
clusterName, hostName, hostName, "bin/run-bookie.sh", BOOKIE_PORT, INVALID_PORT);
+ tailContainerLog();
}
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
index 9104390..62f59c3 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java
@@ -27,7 +27,6 @@ import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.schema.AvroSchema;
@@ -35,6 +34,7 @@ import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.apache.pulsar.client.impl.schema.KeyValueSchemaImpl;
import org.apache.pulsar.client.impl.schema.ProtobufNativeSchema;
import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.KeyValueEncodingType;
import org.apache.pulsar.common.schema.SchemaType;
@@ -56,6 +56,7 @@ public class TestBasicPresto extends TestPulsarSQLBase {
private void setupPresto() throws Exception {
log.info("[TestBasicPresto] setupPresto...");
pulsarCluster.startPrestoWorker();
+ initJdbcConnection();
}
private void teardownPresto() {
@@ -161,31 +162,26 @@ public class TestBasicPresto extends TestPulsarSQLBase {
boolean useNsOffloadPolices,
Schema schema,
CompressionType compressionType) throws Exception {
- @Cleanup
- PulsarClient pulsarClient = PulsarClient.builder()
- .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
- .build();
if (schema.getSchemaInfo().getName().equals(Schema.BYTES.getSchemaInfo().getName())) {
- prepareDataForBytesSchema(pulsarClient, topicName, isBatch, compressionType);
+ prepareDataForBytesSchema(topicName, isBatch, compressionType);
} else if (schema.getSchemaInfo().getName().equals(Schema.BYTEBUFFER.getSchemaInfo().getName())) {
- prepareDataForByteBufferSchema(pulsarClient, topicName, isBatch, compressionType);
+ prepareDataForByteBufferSchema(topicName, isBatch, compressionType);
} else if (schema.getSchemaInfo().getType().equals(SchemaType.STRING)) {
- prepareDataForStringSchema(pulsarClient, topicName, isBatch, compressionType);
+ prepareDataForStringSchema(topicName, isBatch, compressionType);
} else if (schema.getSchemaInfo().getType().equals(SchemaType.JSON)
|| schema.getSchemaInfo().getType().equals(SchemaType.AVRO)) {
- prepareDataForStructSchema(pulsarClient, topicName, isBatch, schema, compressionType);
+ prepareDataForStructSchema(topicName, isBatch, schema, compressionType);
} else if (schema.getSchemaInfo().getType().equals(SchemaType.PROTOBUF_NATIVE)) {
- prepareDataForProtobufNativeSchema(pulsarClient, topicName, isBatch, schema, compressionType);
+ prepareDataForProtobufNativeSchema(topicName, isBatch, schema, compressionType);
} else if (schema.getSchemaInfo().getType().equals(SchemaType.KEY_VALUE)) {
- prepareDataForKeyValueSchema(pulsarClient, topicName, schema, compressionType);
+ prepareDataForKeyValueSchema(topicName, schema, compressionType);
}
return NUM_OF_STOCKS;
}
- private void prepareDataForBytesSchema(PulsarClient pulsarClient,
- TopicName topicName,
+ private void prepareDataForBytesSchema(TopicName topicName,
boolean isBatch,
CompressionType compressionType) throws PulsarClientException {
@Cleanup
@@ -201,8 +197,7 @@ public class TestBasicPresto extends TestPulsarSQLBase {
producer.flush();
}
- private void prepareDataForByteBufferSchema(PulsarClient pulsarClient,
- TopicName topicName,
+ private void prepareDataForByteBufferSchema(TopicName topicName,
boolean isBatch,
CompressionType compressionType) throws PulsarClientException {
@Cleanup
@@ -218,8 +213,7 @@ public class TestBasicPresto extends TestPulsarSQLBase {
producer.flush();
}
- private void prepareDataForStringSchema(PulsarClient pulsarClient,
- TopicName topicName,
+ private void prepareDataForStringSchema(TopicName topicName,
boolean isBatch,
CompressionType compressionType) throws PulsarClientException {
@Cleanup
@@ -235,8 +229,7 @@ public class TestBasicPresto extends TestPulsarSQLBase {
producer.flush();
}
- private void prepareDataForStructSchema(PulsarClient pulsarClient,
- TopicName topicName,
+ private void prepareDataForStructSchema(TopicName topicName,
boolean isBatch,
Schema<Stock> schema,
CompressionType compressionType) throws Exception {
@@ -254,8 +247,7 @@ public class TestBasicPresto extends TestPulsarSQLBase {
producer.flush();
}
- private void prepareDataForProtobufNativeSchema(PulsarClient pulsarClient,
- TopicName topicName,
+ private void prepareDataForProtobufNativeSchema(TopicName topicName,
boolean isBatch,
Schema<StockProtoMessage.Stock> schema,
CompressionType compressionType) throws Exception {
@@ -274,8 +266,7 @@ public class TestBasicPresto extends TestPulsarSQLBase {
producer.flush();
}
- private void prepareDataForKeyValueSchema(PulsarClient pulsarClient,
- TopicName topicName,
+ private void prepareDataForKeyValueSchema(TopicName topicName,
Schema<KeyValue<Stock, Stock>> schema,
CompressionType compressionType) throws Exception {
@Cleanup
@@ -342,4 +333,33 @@ public class TestBasicPresto extends TestPulsarSQLBase {
}
}
+ @Test(timeOut = 1000 * 30)
+ public void testQueueBigEntry() throws Exception {
+ String tableName = "big_data_" + randomName(5);
+ String topic = "persistent://public/default/" + tableName;
+
+ @Cleanup
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .enableBatching(false)
+ .create();
+
+ // Make sure that the data length bigger than the default maxMessageSize
+ int dataLength = Commands.DEFAULT_MAX_MESSAGE_SIZE + 2 * 1024 * 1024;
+ Assert.assertTrue(dataLength < pulsarCluster.getSpec().maxMessageSize());
+ byte[] data = new byte[dataLength];
+ for (int i = 0; i < dataLength; i++) {
+ data[i] = 'a';
+ }
+
+ int messageCnt = 5;
+ log.info("start produce big entry data, data length: {}", dataLength);
+ for (int i = 0 ; i < messageCnt; ++i) {
+ producer.newMessage().value(data).send();
+ }
+
+ int count = selectCount("public/default", tableName);
+ Assert.assertEquals(count, messageCnt);
+ }
+
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java
index 881dbe4..4c129fa 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPrestoQueryTieredStorage.java
@@ -29,7 +29,6 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.CompressionType;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.impl.MessageIdImpl;
@@ -40,8 +39,6 @@ import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.tests.integration.containers.S3Container;
import org.testcontainers.shaded.org.apache.commons.lang.StringUtils;
import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -89,6 +86,7 @@ public class TestPrestoQueryTieredStorage extends TestPulsarSQLBase {
String offloadProperties = getOffloadProperties(BUCKET, null, ENDPOINT);
pulsarCluster.startPrestoWorker(OFFLOAD_DRIVER, offloadProperties);
pulsarCluster.startPrestoFollowWorkers(1, OFFLOAD_DRIVER, offloadProperties);
+ initJdbcConnection();
}
private String getOffloadProperties(String bucket, String region, String endpoint) {
@@ -136,11 +134,6 @@ public class TestPrestoQueryTieredStorage extends TestPulsarSQLBase {
Schema schema,
CompressionType compressionType) throws Exception {
@Cleanup
- PulsarClient pulsarClient = PulsarClient.builder()
- .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
- .build();
-
- @Cleanup
Consumer<Stock> consumer = pulsarClient.newConsumer(JSONSchema.of(Stock.class))
.topic(topicName.toString())
.subscriptionName("test")
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLBase.java
index 54b82fb..5e85e94 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestPulsarSQLBase.java
@@ -21,8 +21,6 @@ package org.apache.pulsar.tests.integration.presto;
import static org.assertj.core.api.Assertions.assertThat;
import com.google.common.base.Stopwatch;
-import java.sql.Connection;
-import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
@@ -188,9 +186,6 @@ public class TestPulsarSQLBase extends PulsarSQLTestSuite {
);
// test predicate pushdown
- String url = String.format("jdbc:presto://%s", pulsarCluster.getPrestoWorkerContainer().getUrl());
- Connection connection = DriverManager.getConnection(url, "test", null);
-
String query = String.format("select * from pulsar" +
".\"%s\".\"%s\" order by __publish_time__", namespace, topic);
log.info("Executing query: {}", query);
@@ -259,11 +254,7 @@ public class TestPulsarSQLBase extends PulsarSQLTestSuite {
log.info("Executing query: result for topic {} returnedTimestamps size: {}", topic, returnedTimestamps.size());
assertThat(returnedTimestamps.size()).isEqualTo(0);
- query = String.format("select count(*) from pulsar.\"%s\".\"%s\"", namespace, topic);
- log.info("Executing query: {}", query);
- res = connection.createStatement().executeQuery(query);
- res.next();
- int count = res.getInt("_col0");
+ int count = selectCount(namespace, topic);
assertThat(count).isGreaterThan(messageNum - 2);
}
@@ -296,5 +287,12 @@ public class TestPulsarSQLBase extends PulsarSQLTestSuite {
}
+ protected int selectCount(String namespace, String tableName) throws SQLException {
+ String query = String.format("select count(*) from pulsar.\"%s\".\"%s\"", namespace, tableName);
+ log.info("Executing count query: {}", query);
+ ResultSet res = connection.createStatement().executeQuery(query);
+ res.next();
+ return res.getInt("_col0");
+ }
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarSQLTestSuite.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarSQLTestSuite.java
index 9ed7335..762fff7 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarSQLTestSuite.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/suites/PulsarSQLTestSuite.java
@@ -18,13 +18,22 @@
*/
package org.apache.pulsar.tests.integration.suites;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.tests.integration.containers.BrokerContainer;
import org.apache.pulsar.tests.integration.containers.S3Container;
import org.apache.pulsar.tests.integration.topologies.PulsarClusterSpec;
+/**
+ * Pulsar SQL test suite.
+ */
@Slf4j
public abstract class PulsarSQLTestSuite extends PulsarTestSuite {
@@ -33,11 +42,15 @@ public abstract class PulsarSQLTestSuite extends PulsarTestSuite {
public static final String BUCKET = "pulsar-integtest";
public static final String ENDPOINT = "http://" + S3Container.NAME + ":9090";
+ protected Connection connection = null;
+ protected PulsarClient pulsarClient = null;
+
@Override
protected PulsarClusterSpec.PulsarClusterSpecBuilder beforeSetupCluster(String clusterName, PulsarClusterSpec.PulsarClusterSpecBuilder specBuilder) {
specBuilder.queryLastMessage(true);
specBuilder.clusterName("pulsar-sql-test");
specBuilder.numBrokers(1);
+ specBuilder.maxMessageSize(2 * Commands.DEFAULT_MAX_MESSAGE_SIZE);
return super.beforeSetupCluster(clusterName, specBuilder);
}
@@ -55,4 +68,43 @@ public abstract class PulsarSQLTestSuite extends PulsarTestSuite {
}
}
+ @Override
+ public void setupCluster() throws Exception {
+ super.setupCluster();
+ pulsarClient = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
+ .build();
+ }
+
+ protected void initJdbcConnection() throws SQLException {
+ if (pulsarCluster.getPrestoWorkerContainer() == null) {
+ log.error("The presto work container isn't exist.");
+ return;
+ }
+ String url = String.format("jdbc:presto://%s", pulsarCluster.getPrestoWorkerContainer().getUrl());
+ connection = DriverManager.getConnection(url, "test", null);
+ }
+
+ @Override
+ public void tearDownCluster() throws Exception {
+ close();
+ super.tearDownCluster();
+ }
+
+ protected void close() {
+ if (connection != null) {
+ try {
+ connection.close();
+ } catch (SQLException e) {
+ log.error("Failed to close sql connection.", e);
+ }
+ }
+ if (pulsarClient != null) {
+ try {
+ pulsarClient.close();
+ } catch (PulsarClientException e) {
+ log.error("Failed to close pulsar client.", e);
+ }
+ }
+ }
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
index 5702370..f06dda6 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarCluster.java
@@ -72,6 +72,7 @@ public class PulsarCluster {
return new PulsarCluster(spec);
}
+ @Getter
private final PulsarClusterSpec spec;
@Getter
@@ -149,6 +150,7 @@ public class PulsarCluster {
.withEnv("journalMaxGroupWaitMSec", "0")
.withEnv("clusterName", clusterName)
.withEnv("diskUsageThreshold", "0.99")
+ .withEnv("nettyMaxFrameSizeBytes", "" + spec.maxMessageSize)
)
);
@@ -165,7 +167,8 @@ public class PulsarCluster {
.withEnv("brokerServiceCompactionMonitorIntervalInSeconds", "1")
// used in s3 tests
.withEnv("AWS_ACCESS_KEY_ID", "accesskey")
- .withEnv("AWS_SECRET_KEY", "secretkey");
+ .withEnv("AWS_SECRET_KEY", "secretkey")
+ .withEnv("maxMessageSize", "" + spec.maxMessageSize);
if (spec.queryLastMessage) {
brokerContainer.withEnv("bookkeeperExplicitLacIntervalInMills", "10");
brokerContainer.withEnv("bookkeeperUseV2WireProtocol", "false");
@@ -419,6 +422,7 @@ public class PulsarCluster {
.withEnv("zookeeperServers", ZKContainer.NAME + ":" + ZKContainer.ZK_PORT)
.withEnv("pulsar.zookeeper-uri", ZKContainer.NAME + ":" + ZKContainer.ZK_PORT)
.withEnv("pulsar.web-service-url", "http://pulsar-broker-0:8080")
+ .withEnv("SQL_PREFIX_pulsar.max-message-size", "" + spec.maxMessageSize)
.withClasspathResourceMapping(
resourcePath, "/pulsar/conf/presto/config.properties", BindMode.READ_WRITE);
if (spec.queryLastMessage) {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java
index 9dcfcfb..eed6042 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarClusterSpec.java
@@ -29,6 +29,7 @@ import lombok.Setter;
import lombok.Singular;
import lombok.experimental.Accessors;
+import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.tests.integration.containers.PulsarContainer;
import org.testcontainers.containers.GenericContainer;
@@ -152,4 +153,7 @@ public class PulsarClusterSpec {
* Specify mount files.
*/
Map<String, String> brokerMountFiles;
+
+ @Default
+ int maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE;
}
[pulsar] 05/09: fix a typo in UnAckedMessageTracker (#12467)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4884fb7d2bc785e0197234b4ecf0a7fc1d00bd1e
Author: ZhangJian He <sh...@gmail.com>
AuthorDate: Sun Oct 24 12:32:29 2021 +0800
fix a typo in UnAckedMessageTracker (#12467)
(cherry picked from commit 52336c8d0253ca16d51105abc180dcc81e65a5e3)
---
.../org/apache/pulsar/client/impl/NegativeAcksTracker.java | 4 ++--
.../org/apache/pulsar/client/impl/UnAckedMessageTracker.java | 11 +++++------
2 files changed, 7 insertions(+), 8 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
index bbdd786..16cfa0c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/NegativeAcksTracker.java
@@ -28,7 +28,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
-import static org.apache.pulsar.client.impl.UnAckedMessageTracker.addChunkedMessageIdsAndRemoveFromSequnceMap;
+import static org.apache.pulsar.client.impl.UnAckedMessageTracker.addChunkedMessageIdsAndRemoveFromSequenceMap;
class NegativeAcksTracker {
@@ -63,7 +63,7 @@ class NegativeAcksTracker {
long now = System.nanoTime();
nackedMessages.forEach((msgId, timestamp) -> {
if (timestamp < now) {
- addChunkedMessageIdsAndRemoveFromSequnceMap(msgId, messagesToRedeliver, this.consumer);
+ addChunkedMessageIdsAndRemoveFromSequenceMap(msgId, messagesToRedeliver, this.consumer);
messagesToRedeliver.add(msgId);
}
});
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
index db616f2..ff07156 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java
@@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.ArrayDeque;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
@@ -130,7 +131,7 @@ public class UnAckedMessageTracker implements Closeable {
if (!headPartition.isEmpty()) {
log.warn("[{}] {} messages have timed-out", consumerBase, headPartition.size());
headPartition.forEach(messageId -> {
- addChunkedMessageIdsAndRemoveFromSequnceMap(messageId, messageIds, consumerBase);
+ addChunkedMessageIdsAndRemoveFromSequenceMap(messageId, messageIds, consumerBase);
messageIds.add(messageId);
messageIdPartitionMap.remove(messageId);
});
@@ -150,14 +151,12 @@ public class UnAckedMessageTracker implements Closeable {
}, this.tickDurationInMs, TimeUnit.MILLISECONDS);
}
- public static void addChunkedMessageIdsAndRemoveFromSequnceMap(MessageId messageId, Set<MessageId> messageIds,
- ConsumerBase<?> consumerBase) {
+ public static void addChunkedMessageIdsAndRemoveFromSequenceMap(MessageId messageId, Set<MessageId> messageIds,
+ ConsumerBase<?> consumerBase) {
if (messageId instanceof MessageIdImpl) {
MessageIdImpl[] chunkedMsgIds = consumerBase.unAckedChunkedMessageIdSequenceMap.get((MessageIdImpl) messageId);
if (chunkedMsgIds != null && chunkedMsgIds.length > 0) {
- for (MessageIdImpl msgId : chunkedMsgIds) {
- messageIds.add(msgId);
- }
+ Collections.addAll(messageIds, chunkedMsgIds);
}
consumerBase.unAckedChunkedMessageIdSequenceMap.remove((MessageIdImpl) messageId);
}
[pulsar] 06/09: Optimize the code: remove extra spaces (#12470)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 83d941f238b488b42618e3178ac298a1e6fa433c
Author: chenlin <15...@qq.com>
AuthorDate: Sun Oct 24 12:28:29 2021 +0800
Optimize the code: remove extra spaces (#12470)
(cherry picked from commit 95e6a0edd74d51b2275ae24083276259d7a92f91)
---
.../src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index d2b4f2a..87e4ee9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -717,7 +717,7 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
protected void onAckTimeoutSend(Set<MessageId> messageIds) {
if (interceptors != null) {
- interceptors. onAckTimeoutSend(this, messageIds);
+ interceptors.onAckTimeoutSend(this, messageIds);
}
}