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/12/21 01:33:36 UTC
[pulsar] branch branch-2.9 updated (626b1d3 -> 8e68926)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 626b1d3 [pulsar-perf]Support listenerThreads configuration. (#12892)
new fbadbd0 [Issue 12897] Fix flaky test in testReplicatorProducerName (#12898)
new 4b78923 [Transaction] Add transaction perf (#11933)
new 4954c09 [Java Client] Avoid IllegalStateException in ClientCnx debug logs (#12899)
new 96238e0 [broker] Fix topic policy listener deleted by mistake. (#12904)
new ca6540e [Broker] Correct param of delete method for v1 topic (#12936)
new 29e12dd [CI] Replace deprecated "adopt" distribution with "temurin" (#12945)
new f13e9e3 Use JDK default security provider when Conscrypt isn't available (#12938)
new 13b07fd [function] pulsar admin exposes secrets for function (#12950)
new 8113c17 Fix wrong isEmpty method of ConcurrentOpenLongPairRangeSet (#12953)
new a5c3466 [c++] Define and expose PULSAR_VERSION macro (#12769)
new 73863f8 Fix NPE in checkSubscriptionTypesEnable (#12961)
new da973ca feat(cli): support autorecovery service in pulsar cli (#12985)
new 71ba8ef Fix update ledger list to znode version mismatch failed, ledger not delete (#12015)
new 6b9d02c optimize if statement (#12741)
new 5a84ae7 Don't create AvroData for each KafkaSourceRecord (#12859)
new ab57a6a Use sendAsync instead of send when produce message to retry topic. (#12946)
new db1a34c fix fix-13004 Race condition in ResourceLockImpl#revalidate (#13006)
new 8e68926 add missed import (#13037)
The 18 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:
.github/workflows/ci-build-macos.yaml | 2 +-
.github/workflows/ci-cpp.yaml | 2 +-
.../ci-integration-backwards-compatibility.yaml | 2 +-
.github/workflows/ci-integration-cli.yaml | 2 +-
.github/workflows/ci-integration-function.yaml | 2 +-
.github/workflows/ci-integration-messaging.yaml | 2 +-
.github/workflows/ci-integration-process.yaml | 2 +-
.../workflows/ci-integration-pulsar-io-ora.yaml | 2 +-
.github/workflows/ci-integration-pulsar-io.yaml | 2 +-
.github/workflows/ci-integration-schema.yaml | 2 +-
.github/workflows/ci-integration-sql.yaml | 2 +-
.github/workflows/ci-integration-standalone.yaml | 2 +-
.github/workflows/ci-integration-thread.yaml | 2 +-
.../ci-integration-tiered-filesystem.yaml | 2 +-
.../workflows/ci-integration-tiered-jcloud.yaml | 2 +-
.github/workflows/ci-integration-transaction.yaml | 2 +-
.github/workflows/ci-license.yaml | 2 +-
.github/workflows/ci-maven-cache-update.yaml | 2 +-
.github/workflows/ci-owasp-dependency-check.yaml | 2 +-
.github/workflows/ci-pulsar-website-build.yaml | 2 +-
.github/workflows/ci-shade-test.yaml | 2 +-
.github/workflows/ci-unit-broker-broker-gp1.yaml | 2 +-
.github/workflows/ci-unit-broker-broker-gp2.yaml | 2 +-
.github/workflows/ci-unit-broker-client-api.yaml | 2 +-
.github/workflows/ci-unit-broker-client-impl.yaml | 2 +-
.github/workflows/ci-unit-broker-jdk8.yaml | 2 +-
.github/workflows/ci-unit-broker-other.yaml | 2 +-
.github/workflows/ci-unit-proxy.yaml | 2 +-
.github/workflows/ci-unit.yaml | 2 +-
bin/pulsar | 4 +
bin/pulsar-daemon | 4 +
bin/pulsar-perf | 3 +
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 30 +-
.../pulsar/broker/admin/v1/PersistentTopics.java | 10 +-
.../broker/service/persistent/PersistentTopic.java | 23 +-
.../pulsar/broker/service/ReplicatorTest.java | 10 +-
.../SystemTopicBasedTopicPoliciesServiceTest.java | 24 +
.../java/org/apache/pulsar/schema/SchemaTest.java | 90 +++
pulsar-client-cpp/.gitignore | 3 +
pulsar-client-cpp/CMakeLists.txt | 4 +
.../include/pulsar/c/version.h | 5 +-
pulsar-client-cpp/lib/CMakeLists.txt | 2 +-
pulsar-client-cpp/lib/Commands.cc | 6 +-
pulsar-client-cpp/lib/HTTPLookupService.cc | 2 +-
pulsar-client-cpp/lib/HTTPLookupService.h | 2 +-
.../lib/{Version.h => VersionInternal.h} | 10 +-
.../templates/Version.h.in | 9 +-
.../tests/{LoggerTest.cc => VersionTest.cc} | 15 +-
.../org/apache/pulsar/admin/cli/CliCommand.java | 1 +
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 18 +-
.../java/org/apache/pulsar/admin/cli/CmdSinks.java | 6 +-
.../org/apache/pulsar/admin/cli/CmdSources.java | 6 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 10 +-
.../apache/pulsar/client/impl/ConsumerBase.java | 7 +-
.../apache/pulsar/client/impl/ConsumerImpl.java | 10 +-
.../pulsar/client/impl/schema/ByteBufSchema.java | 6 +-
.../org/apache/pulsar/common/naming/TopicName.java | 8 +
.../pulsar/common/tls/TlsHostnameVerifier.java | 4 +-
.../apache/pulsar/common/util/SecurityUtility.java | 11 +-
.../ConcurrentOpenLongPairRangeSet.java | 12 +-
.../ConcurrentOpenLongPairRangeSetTest.java | 13 +
.../pulsar/functions/worker/FunctionActioner.java | 7 +-
.../org/apache/pulsar/io/flume/FlumeConnector.java | 5 +-
.../io/kafka/connect/KafkaConnectSource.java | 5 +-
.../coordination/impl/ResourceLockImpl.java | 28 +
.../testclient/CmdGenerateDocumentation.java | 1 +
.../pulsar/testclient/PerformanceConsumer.java | 252 ++++++--
.../pulsar/testclient/PerformanceProducer.java | 160 ++++-
.../pulsar/testclient/PerformanceReader.java | 2 +-
.../pulsar/testclient/PerformanceTransaction.java | 707 +++++++++++++++++++++
.../pulsar/testclient/utils/PerformanceUtils.java | 59 ++
.../testclient/PerformanceTransactionTest.java | 240 +++++++
...ject-version.py => gen-pulsar-version-macro.py} | 8 +-
73 files changed, 1703 insertions(+), 197 deletions(-)
copy pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/package-info.java => pulsar-client-cpp/include/pulsar/c/version.h (94%)
rename pulsar-client-cpp/lib/{Version.h => VersionInternal.h} (82%)
copy pulsar-common/src/main/java/io/airlift/compress/zstd/package-info.java => pulsar-client-cpp/templates/Version.h.in (86%)
copy pulsar-client-cpp/tests/{LoggerTest.cc => VersionTest.cc} (81%)
create mode 100644 pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
create mode 100644 pulsar-testclient/src/main/java/org/apache/pulsar/testclient/utils/PerformanceUtils.java
create mode 100644 pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceTransactionTest.java
copy src/{get-project-version.py => gen-pulsar-version-macro.py} (82%)
[pulsar] 17/18: fix fix-13004 Race condition in ResourceLockImpl#revalidate (#13006)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit db1a34cfe708c182b9fa54515e0a26c3c042163b
Author: JiangHaiting <ji...@apache.org>
AuthorDate: Tue Nov 30 01:32:22 2021 +0800
fix fix-13004 Race condition in ResourceLockImpl#revalidate (#13006)
(cherry picked from commit bb2c9345bc42d61090897757a9f7435c138c8ee3)
---
.../coordination/impl/ResourceLockImpl.java | 28 ++++++++++++++++++++++
1 file changed, 28 insertions(+)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
index 677ace7..ec9d3c4 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/coordination/impl/ResourceLockImpl.java
@@ -44,6 +44,7 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
private long version;
private final CompletableFuture<Void> expiredFuture;
private boolean revalidateAfterReconnection = false;
+ private CompletableFuture<Void> revalidateFuture;
private enum State {
Init,
@@ -144,6 +145,9 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
// Simple operation of acquiring the lock with no retries, or checking for the lock content
private CompletableFuture<Void> acquireWithNoRevalidation(T newValue) {
+ if (log.isDebugEnabled()) {
+ log.debug("acquireWithNoRevalidation,newValue={},version={}", newValue, version);
+ }
byte[] payload;
try {
payload = serde.serialize(path, newValue);
@@ -212,6 +216,30 @@ public class ResourceLockImpl<T> implements ResourceLock<T> {
}
synchronized CompletableFuture<Void> revalidate(T newValue) {
+ if (revalidateFuture == null || revalidateFuture.isDone()) {
+ revalidateFuture = doRevalidate(newValue);
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Previous revalidating is not finished while revalidate newValue={}, value={}, version={}",
+ newValue, value, version);
+ }
+ CompletableFuture<Void> newFuture = new CompletableFuture<>();
+ revalidateFuture.whenComplete((unused, throwable) -> {
+ doRevalidate(newValue).thenRun(() -> newFuture.complete(null))
+ .exceptionally(throwable1 -> {
+ newFuture.completeExceptionally(throwable1);
+ return null;
+ });
+ });
+ revalidateFuture = newFuture;
+ }
+ return revalidateFuture;
+ }
+
+ private synchronized CompletableFuture<Void> doRevalidate(T newValue) {
+ if (log.isDebugEnabled()) {
+ log.debug("doRevalidate with newValue={}, version={}", newValue, version);
+ }
return store.get(path)
.thenCompose(optGetResult -> {
if (!optGetResult.isPresent()) {
[pulsar] 12/18: feat(cli): support autorecovery service in pulsar cli (#12985)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit da973ca0a1e1221b46db546790e8469e630125c5
Author: Eric Shen <er...@outlook.com>
AuthorDate: Fri Nov 26 20:08:51 2021 -0600
feat(cli): support autorecovery service in pulsar cli (#12985)
### Motivation
Autorecovery service will be shutdown if the zk session expired and then will lead the bk service shutdown together. So, in the production environment, it is recommand to deploy autorecovery service seperately but currently pulsar doesn't support it.
### Modifications
Added the autorecovery service in pulsar cli
Added the autorecovery service in pulsar-daemon cli
(cherry picked from commit f192209af58e86a2c6da4130956b172b2d1ddc13)
---
bin/pulsar | 4 ++++
bin/pulsar-daemon | 4 ++++
2 files changed, 8 insertions(+)
diff --git a/bin/pulsar b/bin/pulsar
index 3ec65f3..0640a94 100755
--- a/bin/pulsar
+++ b/bin/pulsar
@@ -140,6 +140,7 @@ where command is one of:
sql-worker Run a sql worker server
sql Run sql CLI
standalone Run a broker server with local bookies and local zookeeper
+ autorecovery Run an autorecovery service
initialize-cluster-metadata One-time metadata initialization
delete-cluster-metadata Delete a cluster's metadata
@@ -343,6 +344,9 @@ elif [ $COMMAND == "functions-worker" ]; then
elif [ $COMMAND == "standalone" ]; then
PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-standalone.log"}
exec $JAVA $LOG4J2_SHUTDOWN_HOOK_DISABLED $OPTS ${ZK_OPTS} -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarStandaloneStarter --config $PULSAR_STANDALONE_CONF $@
+elif [ ${COMMAND} == "autorecovery" ]; then
+ PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-autorecovery.log"}
+ exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.replication.AutoRecoveryMain --conf $PULSAR_BOOKKEEPER_CONF $@
elif [ $COMMAND == "initialize-cluster-metadata" ]; then
exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataSetup $@
elif [ $COMMAND == "delete-cluster-metadata" ]; then
diff --git a/bin/pulsar-daemon b/bin/pulsar-daemon
index f018bc4..8bbc628 100755
--- a/bin/pulsar-daemon
+++ b/bin/pulsar-daemon
@@ -30,6 +30,7 @@ where command is one of:
functions-worker Run a functions worker server
standalone Run a standalone Pulsar service
proxy Run a Proxy Pulsar service
+ autorecovery Run an autorecovery service
where argument is one of:
-force (accepted only with stop command): Decides whether to stop the server forcefully if not stopped by normal shutdown
@@ -102,6 +103,9 @@ case $command in
(proxy)
echo "doing $startStop $command ..."
;;
+ (autorecovery)
+ echo "doing $startStop $command ..."
+ ;;
(*)
echo "Error: unknown service name $command"
usage
[pulsar] 05/18: [Broker] Correct param of delete method for v1 topic (#12936)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ca6540ed9272c881de533405ccfce945b9d4d96d
Author: Ruguo Yu <ji...@163.com>
AuthorDate: Wed Nov 24 08:37:18 2021 +0800
[Broker] Correct param of delete method for v1 topic (#12936)
(cherry picked from commit e6d9df81c446870107dbb8d8e454b11b71cc9255)
---
.../pulsar/broker/admin/v1/PersistentTopics.java | 10 ++-
.../java/org/apache/pulsar/schema/SchemaTest.java | 90 ++++++++++++++++++++++
2 files changed, 96 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
index 9ef87f4..c0c4b48 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
@@ -264,10 +264,11 @@ public class PersistentTopics extends PersistentTopicsBase {
@PathParam("property") String property, @PathParam("cluster") String cluster,
@PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
@QueryParam("force") @DefaultValue("false") boolean force,
- @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @QueryParam("deleteSchema") @DefaultValue("false") boolean deleteSchema) {
try {
validateTopicName(property, cluster, namespace, encodedTopic);
- internalDeletePartitionedTopic(asyncResponse, authoritative, force, false);
+ internalDeletePartitionedTopic(asyncResponse, authoritative, force, deleteSchema);
} catch (WebApplicationException wae) {
asyncResponse.resume(wae);
} catch (Exception e) {
@@ -302,9 +303,10 @@ public class PersistentTopics extends PersistentTopicsBase {
public void deleteTopic(@PathParam("property") String property, @PathParam("cluster") String cluster,
@PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
@QueryParam("force") @DefaultValue("false") boolean force,
- @QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
+ @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
+ @QueryParam("deleteSchema") @DefaultValue("false") boolean deleteSchema) {
validateTopicName(property, cluster, namespace, encodedTopic);
- internalDeleteTopic(authoritative, force);
+ internalDeleteTopic(authoritative, force, deleteSchema);
}
@GET
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
index db56fbc..45f12ab 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java
@@ -725,6 +725,96 @@ public class SchemaTest extends MockedPulsarServiceBaseTest {
}
@Test
+ public void testDeleteTopicAndSchemaForV1() throws Exception {
+ final String tenant = PUBLIC_TENANT;
+ final String cluster = CLUSTER_NAME;
+ final String namespace = "test-namespace-" + randomName(16);
+ final String topicOne = "not-partitioned-topic";
+ final String topic2 = "persistent://" + tenant + "/" + cluster + "/" + namespace + "/partitioned-topic";
+
+ // persistent, not-partitioned v1/topic
+ final String topic1 = TopicName.get(
+ TopicDomain.persistent.value(),
+ tenant,
+ cluster,
+ namespace,
+ topicOne).toString();
+
+ // persistent, partitioned v1/topic
+ admin.topics().createPartitionedTopic(topic2, 1);
+
+ @Cleanup
+ Producer<Schemas.PersonOne> p1_1 = pulsarClient.newProducer(Schema.JSON(Schemas.PersonOne.class))
+ .topic(topic1)
+ .create();
+
+ @Cleanup
+ Producer<Schemas.PersonThree> p1_2 = pulsarClient.newProducer(Schema.JSON(Schemas.PersonThree.class))
+ .topic(topic1)
+ .create();
+ @Cleanup
+ Producer<Schemas.PersonThree> p2_1 = pulsarClient.newProducer(Schema.JSON(Schemas.PersonThree.class))
+ .topic(topic2)
+ .create();
+
+ List<CompletableFuture<SchemaRegistry.SchemaAndMetadata>> schemaFutures1 =
+ this.getPulsar().getSchemaRegistryService().getAllSchemas(TopicName.get(topic1).getSchemaName()).get();
+ FutureUtil.waitForAll(schemaFutures1).get();
+ List<SchemaRegistry.SchemaAndMetadata> schemas1 = schemaFutures1.stream().map(future -> {
+ try {
+ return future.get();
+ } catch (Exception e) {
+ return null;
+ }
+ }).collect(Collectors.toList());
+ assertEquals(schemas1.size(), 2);
+ for (SchemaRegistry.SchemaAndMetadata schema : schemas1) {
+ assertNotNull(schema);
+ }
+
+ List<CompletableFuture<SchemaRegistry.SchemaAndMetadata>> schemaFutures2 =
+ this.getPulsar().getSchemaRegistryService().getAllSchemas(TopicName.get(topic2).getSchemaName()).get();
+ FutureUtil.waitForAll(schemaFutures2).get();
+ List<SchemaRegistry.SchemaAndMetadata> schemas2 = schemaFutures2.stream().map(future -> {
+ try {
+ return future.get();
+ } catch (Exception e) {
+ return null;
+ }
+ }).collect(Collectors.toList());
+ assertEquals(schemas2.size(), 1);
+ for (SchemaRegistry.SchemaAndMetadata schema : schemas2) {
+ assertNotNull(schema);
+ }
+
+ // not-force and not-delete-schema when delete topic
+ try {
+ admin.topics().delete(topic1, false, false);
+ fail();
+ } catch (Exception e) {
+ assertTrue(e.getMessage().startsWith("Topic has active producers/subscriptions"));
+ }
+ assertEquals(this.getPulsar().getSchemaRegistryService()
+ .trimDeletedSchemaAndGetList(TopicName.get(topic1).getSchemaName()).get().size(), 2);
+ try {
+ admin.topics().deletePartitionedTopic(topic2, false, false);
+ fail();
+ } catch (Exception e) {
+ assertTrue(e.getMessage().startsWith("Topic has active producers/subscriptions"));
+ }
+ assertEquals(this.getPulsar().getSchemaRegistryService()
+ .trimDeletedSchemaAndGetList(TopicName.get(topic2).getSchemaName()).get().size(), 1);
+
+ // force and delete-schema when delete topic
+ admin.topics().delete(topic1, true, true);
+ assertEquals(this.getPulsar().getSchemaRegistryService()
+ .trimDeletedSchemaAndGetList(TopicName.get(topic1).getSchemaName()).get().size(), 0);
+ admin.topics().deletePartitionedTopic(topic2, true, true);
+ assertEquals(this.getPulsar().getSchemaRegistryService()
+ .trimDeletedSchemaAndGetList(TopicName.get(topic2).getSchemaName()).get().size(), 0);
+ }
+
+ @Test
public void testProducerMultipleSchemaMessages() throws Exception {
final String tenant = PUBLIC_TENANT;
final String namespace = "test-namespace-" + randomName(16);
[pulsar] 15/18: Don't create AvroData for each KafkaSourceRecord (#12859)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5a84ae708d61d9b1d0b450227211a81d5ecb33f9
Author: Andrey Yegorov <86...@users.noreply.github.com>
AuthorDate: Mon Nov 29 10:54:48 2021 -0800
Don't create AvroData for each KafkaSourceRecord (#12859)
(cherry picked from commit 792a77ebf841c8cae5ffe8a0cba5bbf1ed64f83c)
---
.../java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java | 5 +----
1 file changed, 1 insertion(+), 4 deletions(-)
diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
index f84aa64..5d30e95 100644
--- a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
+++ b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java
@@ -69,15 +69,12 @@ public class KafkaConnectSource extends AbstractKafkaConnectSource<KeyValue<byte
return record;
}
- private static Map<String, String> PROPERTIES = Collections.emptyMap();
- private static Optional<Long> RECORD_SEQUENCE = Optional.empty();
- private static long FLUSH_TIMEOUT_MS = 2000;
+ private static final AvroData avroData = new AvroData(1000);
private class KafkaSourceRecord extends AbstractKafkaSourceRecord<KeyValue<byte[], byte[]>> implements KVRecord<byte[], byte[]> {
KafkaSourceRecord(SourceRecord srcRecord) {
super(srcRecord);
- AvroData avroData = new AvroData(1000);
byte[] keyBytes = keyConverter.fromConnectData(
srcRecord.topic(), srcRecord.keySchema(), srcRecord.key());
this.key = keyBytes != null ? Optional.of(Base64.getEncoder().encodeToString(keyBytes)) : Optional.empty();
[pulsar] 07/18: Use JDK default security provider when Conscrypt isn't available (#12938)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f13e9e324f9664463bb7499978a38ba0c7d99431
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Thu Nov 25 15:29:24 2021 +0200
Use JDK default security provider when Conscrypt isn't available (#12938)
- fixes issue with ARM64 platform where Conscrypt isn't available
(cherry picked from commit 4f2d52edfbb53f043ed5640ccde694b60707eea3)
---
.../java/org/apache/pulsar/common/util/SecurityUtility.java | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
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 faefc8a..db8f861 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
@@ -116,6 +116,16 @@ public class SecurityUtility {
}
private static Provider loadConscryptProvider() {
+ Class<?> conscryptClazz;
+
+ try {
+ conscryptClazz = Class.forName("org.conscrypt.Conscrypt");
+ conscryptClazz.getMethod("checkAvailability").invoke(null);
+ } catch (Throwable e) {
+ log.warn("Conscrypt isn't available. Using JDK default security provider.", e);
+ return null;
+ }
+
Provider provider;
try {
provider = (Provider) Class.forName(CONSCRYPT_PROVIDER_CLASS).getDeclaredConstructor().newInstance();
@@ -143,7 +153,6 @@ public class SecurityUtility {
// contains the workaround.
try {
HostnameVerifier hostnameVerifier = new TlsHostnameVerifier();
- Class<?> conscryptClazz = Class.forName("org.conscrypt.Conscrypt");
Object wrappedHostnameVerifier = conscryptClazz
.getMethod("wrapHostnameVerifier",
new Class[]{HostnameVerifier.class}).invoke(null, hostnameVerifier);
[pulsar] 06/18: [CI] Replace deprecated "adopt" distribution with "temurin" (#12945)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 29e12dd2b4cb069912d0d06fbbd00853d55c5b80
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Tue Nov 23 19:28:15 2021 +0200
[CI] Replace deprecated "adopt" distribution with "temurin" (#12945)
- see https://github.com/actions/setup-java#supported-distributions
for more details"
(cherry picked from commit 60ef5e983e5e8956bd0b602b5741bd6255c6258a)
---
.github/workflows/ci-build-macos.yaml | 2 +-
.github/workflows/ci-cpp.yaml | 2 +-
.github/workflows/ci-integration-backwards-compatibility.yaml | 2 +-
.github/workflows/ci-integration-cli.yaml | 2 +-
.github/workflows/ci-integration-function.yaml | 2 +-
.github/workflows/ci-integration-messaging.yaml | 2 +-
.github/workflows/ci-integration-process.yaml | 2 +-
.github/workflows/ci-integration-pulsar-io-ora.yaml | 2 +-
.github/workflows/ci-integration-pulsar-io.yaml | 2 +-
.github/workflows/ci-integration-schema.yaml | 2 +-
.github/workflows/ci-integration-sql.yaml | 2 +-
.github/workflows/ci-integration-standalone.yaml | 2 +-
.github/workflows/ci-integration-thread.yaml | 2 +-
.github/workflows/ci-integration-tiered-filesystem.yaml | 2 +-
.github/workflows/ci-integration-tiered-jcloud.yaml | 2 +-
.github/workflows/ci-integration-transaction.yaml | 2 +-
.github/workflows/ci-license.yaml | 2 +-
.github/workflows/ci-maven-cache-update.yaml | 2 +-
.github/workflows/ci-owasp-dependency-check.yaml | 2 +-
.github/workflows/ci-pulsar-website-build.yaml | 2 +-
.github/workflows/ci-shade-test.yaml | 2 +-
.github/workflows/ci-unit-broker-broker-gp1.yaml | 2 +-
.github/workflows/ci-unit-broker-broker-gp2.yaml | 2 +-
.github/workflows/ci-unit-broker-client-api.yaml | 2 +-
.github/workflows/ci-unit-broker-client-impl.yaml | 2 +-
.github/workflows/ci-unit-broker-jdk8.yaml | 2 +-
.github/workflows/ci-unit-broker-other.yaml | 2 +-
.github/workflows/ci-unit-proxy.yaml | 2 +-
.github/workflows/ci-unit.yaml | 2 +-
29 files changed, 29 insertions(+), 29 deletions(-)
diff --git a/.github/workflows/ci-build-macos.yaml b/.github/workflows/ci-build-macos.yaml
index 0236ab3..0d128c4 100644
--- a/.github/workflows/ci-build-macos.yaml
+++ b/.github/workflows/ci-build-macos.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
diff --git a/.github/workflows/ci-cpp.yaml b/.github/workflows/ci-cpp.yaml
index 70f5f6f..ed82379 100644
--- a/.github/workflows/ci-cpp.yaml
+++ b/.github/workflows/ci-cpp.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-backwards-compatibility.yaml b/.github/workflows/ci-integration-backwards-compatibility.yaml
index 3b13b63..59efbab 100644
--- a/.github/workflows/ci-integration-backwards-compatibility.yaml
+++ b/.github/workflows/ci-integration-backwards-compatibility.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-cli.yaml b/.github/workflows/ci-integration-cli.yaml
index cf51dcc..7fb9b95 100644
--- a/.github/workflows/ci-integration-cli.yaml
+++ b/.github/workflows/ci-integration-cli.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-function.yaml b/.github/workflows/ci-integration-function.yaml
index 52b6d8f..1e7e500 100644
--- a/.github/workflows/ci-integration-function.yaml
+++ b/.github/workflows/ci-integration-function.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-messaging.yaml b/.github/workflows/ci-integration-messaging.yaml
index 0a81e4a..68eeda8 100644
--- a/.github/workflows/ci-integration-messaging.yaml
+++ b/.github/workflows/ci-integration-messaging.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-process.yaml b/.github/workflows/ci-integration-process.yaml
index a5523c1..10d3b5e 100644
--- a/.github/workflows/ci-integration-process.yaml
+++ b/.github/workflows/ci-integration-process.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-pulsar-io-ora.yaml b/.github/workflows/ci-integration-pulsar-io-ora.yaml
index 8bb1cba..33b5d11 100644
--- a/.github/workflows/ci-integration-pulsar-io-ora.yaml
+++ b/.github/workflows/ci-integration-pulsar-io-ora.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-pulsar-io.yaml b/.github/workflows/ci-integration-pulsar-io.yaml
index 538e94e2..9b811f2 100644
--- a/.github/workflows/ci-integration-pulsar-io.yaml
+++ b/.github/workflows/ci-integration-pulsar-io.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-schema.yaml b/.github/workflows/ci-integration-schema.yaml
index edeec55..84c02c3 100644
--- a/.github/workflows/ci-integration-schema.yaml
+++ b/.github/workflows/ci-integration-schema.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-sql.yaml b/.github/workflows/ci-integration-sql.yaml
index 53c9968..6b5d7ba 100644
--- a/.github/workflows/ci-integration-sql.yaml
+++ b/.github/workflows/ci-integration-sql.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-standalone.yaml b/.github/workflows/ci-integration-standalone.yaml
index dc35776..17499c4 100644
--- a/.github/workflows/ci-integration-standalone.yaml
+++ b/.github/workflows/ci-integration-standalone.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-thread.yaml b/.github/workflows/ci-integration-thread.yaml
index 0420bae..2133a53 100644
--- a/.github/workflows/ci-integration-thread.yaml
+++ b/.github/workflows/ci-integration-thread.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-tiered-filesystem.yaml b/.github/workflows/ci-integration-tiered-filesystem.yaml
index 89ed30d..413838f 100644
--- a/.github/workflows/ci-integration-tiered-filesystem.yaml
+++ b/.github/workflows/ci-integration-tiered-filesystem.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-tiered-jcloud.yaml b/.github/workflows/ci-integration-tiered-jcloud.yaml
index 3cf661a..5ba8d35 100644
--- a/.github/workflows/ci-integration-tiered-jcloud.yaml
+++ b/.github/workflows/ci-integration-tiered-jcloud.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-integration-transaction.yaml b/.github/workflows/ci-integration-transaction.yaml
index 547066d..cd26248 100644
--- a/.github/workflows/ci-integration-transaction.yaml
+++ b/.github/workflows/ci-integration-transaction.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-license.yaml b/.github/workflows/ci-license.yaml
index f706a1c..ae60426 100644
--- a/.github/workflows/ci-license.yaml
+++ b/.github/workflows/ci-license.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
diff --git a/.github/workflows/ci-maven-cache-update.yaml b/.github/workflows/ci-maven-cache-update.yaml
index b04d286..755f2af 100644
--- a/.github/workflows/ci-maven-cache-update.yaml
+++ b/.github/workflows/ci-maven-cache-update.yaml
@@ -102,7 +102,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ (github.event_name == 'schedule' || steps.changes.outputs.poms == 'true') && steps.cache.outputs.cache-hit != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: Download dependencies
diff --git a/.github/workflows/ci-owasp-dependency-check.yaml b/.github/workflows/ci-owasp-dependency-check.yaml
index 3e95127..301dd25 100644
--- a/.github/workflows/ci-owasp-dependency-check.yaml
+++ b/.github/workflows/ci-owasp-dependency-check.yaml
@@ -53,7 +53,7 @@ jobs:
- name: Set up JDK 11
uses: actions/setup-java@v2
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: run install by skip tests
diff --git a/.github/workflows/ci-pulsar-website-build.yaml b/.github/workflows/ci-pulsar-website-build.yaml
index 58d67ad..1c8658c 100644
--- a/.github/workflows/ci-pulsar-website-build.yaml
+++ b/.github/workflows/ci-pulsar-website-build.yaml
@@ -53,7 +53,7 @@ jobs:
- name: Set up JDK 11
uses: actions/setup-java@v2
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-shade-test.yaml b/.github/workflows/ci-shade-test.yaml
index 8789bf7..1aecf70 100644
--- a/.github/workflows/ci-shade-test.yaml
+++ b/.github/workflows/ci-shade-test.yaml
@@ -69,7 +69,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: clean disk
diff --git a/.github/workflows/ci-unit-broker-broker-gp1.yaml b/.github/workflows/ci-unit-broker-broker-gp1.yaml
index 6defee0..71204e5 100644
--- a/.github/workflows/ci-unit-broker-broker-gp1.yaml
+++ b/.github/workflows/ci-unit-broker-broker-gp1.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: build modules
diff --git a/.github/workflows/ci-unit-broker-broker-gp2.yaml b/.github/workflows/ci-unit-broker-broker-gp2.yaml
index 821157c..6433dd1 100644
--- a/.github/workflows/ci-unit-broker-broker-gp2.yaml
+++ b/.github/workflows/ci-unit-broker-broker-gp2.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: build modules
diff --git a/.github/workflows/ci-unit-broker-client-api.yaml b/.github/workflows/ci-unit-broker-client-api.yaml
index 5bcefdd..699caf0 100644
--- a/.github/workflows/ci-unit-broker-client-api.yaml
+++ b/.github/workflows/ci-unit-broker-client-api.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: build modules
diff --git a/.github/workflows/ci-unit-broker-client-impl.yaml b/.github/workflows/ci-unit-broker-client-impl.yaml
index 21003a1..6841b57 100644
--- a/.github/workflows/ci-unit-broker-client-impl.yaml
+++ b/.github/workflows/ci-unit-broker-client-impl.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: build modules
diff --git a/.github/workflows/ci-unit-broker-jdk8.yaml b/.github/workflows/ci-unit-broker-jdk8.yaml
index ce56ca2..e046085 100644
--- a/.github/workflows/ci-unit-broker-jdk8.yaml
+++ b/.github/workflows/ci-unit-broker-jdk8.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 8
- name: build modules
diff --git a/.github/workflows/ci-unit-broker-other.yaml b/.github/workflows/ci-unit-broker-other.yaml
index 50a87b5..5826244 100644
--- a/.github/workflows/ci-unit-broker-other.yaml
+++ b/.github/workflows/ci-unit-broker-other.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: build modules
diff --git a/.github/workflows/ci-unit-proxy.yaml b/.github/workflows/ci-unit-proxy.yaml
index bc2cfb9..2ffed49 100644
--- a/.github/workflows/ci-unit-proxy.yaml
+++ b/.github/workflows/ci-unit-proxy.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: build modules pulsar-proxy
diff --git a/.github/workflows/ci-unit.yaml b/.github/workflows/ci-unit.yaml
index 37bdfaf..fd553b0 100644
--- a/.github/workflows/ci-unit.yaml
+++ b/.github/workflows/ci-unit.yaml
@@ -68,7 +68,7 @@ jobs:
uses: actions/setup-java@v2
if: ${{ steps.check_changes.outputs.docs_only != 'true' }}
with:
- distribution: 'adopt'
+ distribution: 'temurin'
java-version: 11
- name: run unit test 'OTHER'
[pulsar] 01/18: [Issue 12897] Fix flaky test in testReplicatorProducerName (#12898)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fbadbd07b1fb335ae8db3d4a64ece4fdaf31dbad
Author: JiangHaiting <ji...@apache.org>
AuthorDate: Sun Nov 21 12:49:15 2021 +0800
[Issue 12897] Fix flaky test in testReplicatorProducerName (#12898)
(cherry picked from commit 78c2981a8bc0af4ee883357b99a40e83be8ffa4b)
---
.../java/org/apache/pulsar/broker/service/ReplicatorTest.java | 10 ++++++----
1 file changed, 6 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index 87e2730..488fbfe 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -730,10 +730,12 @@ public class ReplicatorTest extends ReplicatorTestBase {
});
Optional<Topic> topic = pulsar2.getBrokerService().getTopicReference(topicName);
assertTrue(topic.isPresent());
- Set<String> remoteClusters = topic.get().getProducers().values().stream()
- .map(org.apache.pulsar.broker.service.Producer::getRemoteCluster)
- .collect(Collectors.toSet());
- assertTrue(remoteClusters.contains("r1"));
+ Awaitility.await().untilAsserted(() -> {
+ Set<String> remoteClusters = topic.get().getProducers().values().stream()
+ .map(org.apache.pulsar.broker.service.Producer::getRemoteCluster)
+ .collect(Collectors.toSet());
+ assertTrue(remoteClusters.contains("r1"));
+ });
}
@Test(priority = 5, timeOut = 30000)
[pulsar] 02/18: [Transaction] Add transaction perf (#11933)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4b789236c8d7ed1212c7f2007e082f5157bd9df1
Author: Xiangying Meng <55...@users.noreply.github.com>
AuthorDate: Mon Nov 22 16:23:18 2021 +0800
[Transaction] Add transaction perf (#11933)
Master Issue: #https://github.com/apache/pulsar/issues/11765
### Motivation
- When opening a transaction, what is the delay and rate of pulsar sending and consuming messages?
- How does the granularity of the transaction affect the rate at which pulsar produces or consumes messages
- Is the rate of abort and commit transactions different
- Try to keep the functions of ordinary produce perfand consume perf as much as possible
### Modifications
Add PerformanceTransaction.Class
- According to the number of topics set by the user, each topic starts a thread to execute a number of transactions asynchronously.
- If there is a preset transaction value, the production or consumption of this topic is terminated after the corresponding number of transactions is executed, and the production and consumption of all topics are summed up. After the consumption ends, the main thread ends.
- Each consumer can have multiple subscriptions, but each subscription has only one consumer. In order to better compare the rate of production and consumption.
(cherry picked from commit 793a8b95637f534733d2b93b6eedade1399a13a4)
---
bin/pulsar-perf | 3 +
.../testclient/CmdGenerateDocumentation.java | 1 +
.../pulsar/testclient/PerformanceConsumer.java | 252 ++++++--
.../pulsar/testclient/PerformanceProducer.java | 160 ++++-
.../pulsar/testclient/PerformanceTransaction.java | 707 +++++++++++++++++++++
.../pulsar/testclient/utils/PerformanceUtils.java | 59 ++
.../testclient/PerformanceTransactionTest.java | 240 +++++++
7 files changed, 1358 insertions(+), 64 deletions(-)
diff --git a/bin/pulsar-perf b/bin/pulsar-perf
index a2f90b0..cef0cf3 100755
--- a/bin/pulsar-perf
+++ b/bin/pulsar-perf
@@ -87,6 +87,7 @@ Usage: pulsar-perf <command>
where command is one of:
produce Run a producer
consume Run a consumer
+ transaction Run a transaction repeatedly
read Run a topic reader
websocket-producer Run a websocket producer
@@ -164,6 +165,8 @@ if [ "$COMMAND" == "produce" ]; then
exec $JAVA $OPTS org.apache.pulsar.testclient.PerformanceProducer --conf-file $PULSAR_PERFTEST_CONF "$@"
elif [ "$COMMAND" == "consume" ]; then
exec $JAVA $OPTS org.apache.pulsar.testclient.PerformanceConsumer --conf-file $PULSAR_PERFTEST_CONF "$@"
+elif [ "$COMMAND" == "transaction" ]; then
+ exec $JAVA $OPTS org.apache.pulsar.testclient.PerformanceTransaction --conf-file $PULSAR_PERFTEST_CONF "$@"
elif [ "$COMMAND" == "read" ]; then
exec $JAVA $OPTS org.apache.pulsar.testclient.PerformanceReader --conf-file $PULSAR_PERFTEST_CONF "$@"
elif [ "$COMMAND" == "monitor-brokers" ]; then
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java
index c316fc4..29c38ac 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/CmdGenerateDocumentation.java
@@ -64,6 +64,7 @@ public class CmdGenerateDocumentation {
Map<String, Class<?>> cmdClassMap = new LinkedHashMap<>();
cmdClassMap.put("produce", Class.forName("org.apache.pulsar.testclient.PerformanceProducer$Arguments"));
cmdClassMap.put("consume", Class.forName("org.apache.pulsar.testclient.PerformanceConsumer$Arguments"));
+ cmdClassMap.put("transaction", Class.forName("org.apache.pulsar.testclient.PerformanceTransaction$Arguments"));
cmdClassMap.put("read", Class.forName("org.apache.pulsar.testclient.PerformanceReader$Arguments"));
cmdClassMap.put("monitor-brokers", Class.forName("org.apache.pulsar.testclient.BrokerMonitor$Arguments"));
cmdClassMap.put("simulation-client", Class.forName("org.apache.pulsar.testclient.LoadSimulationClient$MainArguments"));
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
index a220c9f..134ffdd 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.testclient;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
@@ -34,7 +35,10 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import org.HdrHistogram.Histogram;
@@ -48,6 +52,7 @@ 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.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
import org.slf4j.Logger;
@@ -67,8 +72,19 @@ public class PerformanceConsumer {
private static final LongAdder totalMessagesReceived = new LongAdder();
private static final LongAdder totalBytesReceived = new LongAdder();
- private static Recorder recorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
- private static Recorder cumulativeRecorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
+ private static final LongAdder totalNumTxnOpenFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenSuccess = new LongAdder();
+
+ private static final LongAdder totalMessageAck = new LongAdder();
+ private static final LongAdder totalMessageAckFailed = new LongAdder();
+ private static final LongAdder messageAck = new LongAdder();
+
+ private static final LongAdder totalEndTxnOpFailNum = new LongAdder();
+ private static final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+
+ private static final Recorder recorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
+ private static final Recorder cumulativeRecorder = new Recorder(TimeUnit.DAYS.toMillis(10), 5);
@Parameters(commandDescription = "Test pulsar consumer performance.")
static class Arguments {
@@ -94,7 +110,7 @@ public class PerformanceConsumer {
@Parameter(names = { "-s", "--subscriber-name" }, description = "Subscriber name prefix", hidden = true)
public String subscriberName;
- @Parameter(names = { "-ss", "--subscriptions" }, description = "A list of subscriptions to consume on (e.g. sub1,sub2)")
+ @Parameter(names = { "-ss", "--subscriptions" }, description = "A list of subscriptions to consume (for example, sub1,sub2)")
public List<String> subscriptions = Collections.singletonList("sub");
@Parameter(names = { "-st", "--subscription-type" }, description = "Subscription type")
@@ -177,7 +193,7 @@ public class PerformanceConsumer {
public long testTime = 0;
@Parameter(names = {"-ioThreads", "--num-io-threads"}, description = "Set the number of threads to be " +
- "used for handling connections to brokers, default is 1 thread")
+ "used for handling connections to brokers. The default value is 1.")
public int ioThreads = 1;
@Parameter(names = {"-lt", "--num-listener-threads"}, description = "Set the number of threads"
@@ -193,6 +209,26 @@ public class PerformanceConsumer {
@Parameter(names = {"-bw", "--busy-wait"}, description = "Enable Busy-Wait on the Pulsar client")
public boolean enableBusyWait = false;
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 10;
+
+ @Parameter(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages acknowledged by a transaction. "
+ + "(After --txn-enable setting to true, -numMessage-perTransaction takes effect")
+ public int numMessagesPerTransaction = 50;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
+
+ @Parameter(names = {"-ntxn"}, description = "The number of opened transactions, 0 means keeping open."
+ + "(After --txn-enable setting to true, -ntxn takes effect.)")
+ public long totalNumTxn = 0;
+
+ @Parameter(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
+ + "setting to true, -abort takes effect)")
+ public boolean isAbortTransaction = false;
+
@Parameter(names = { "--histogram-file" }, description = "HdrHistogram output file")
public String histogramFile = null;
}
@@ -303,43 +339,9 @@ public class PerformanceConsumer {
final RateLimiter limiter = arguments.rate > 0 ? RateLimiter.create(arguments.rate) : null;
long startTime = System.nanoTime();
long testEndTime = startTime + (long) (arguments.testTime * 1e9);
- MessageListener<ByteBuffer> listener = (consumer, msg) -> {
- if (arguments.testTime > 0) {
- if (System.nanoTime() > testEndTime) {
- log.info("------------- DONE (reached the maximum duration: [{} seconds] of consumption) --------------", arguments.testTime);
- printAggregatedStats();
- PerfClientUtils.exit(0);
- }
- }
- if (arguments.numMessages > 0 && totalMessagesReceived.sum() >= arguments.numMessages) {
- log.info("------------- DONE (reached the maximum number: [{}] of consumption) --------------", arguments.numMessages);
- printAggregatedStats();
- PerfClientUtils.exit(0);
- }
- messagesReceived.increment();
- bytesReceived.add(msg.size());
-
- totalMessagesReceived.increment();
- totalBytesReceived.add(msg.size());
-
- if (limiter != null) {
- limiter.acquire();
- }
-
- long latencyMillis = System.currentTimeMillis() - msg.getPublishTime();
- if (latencyMillis >= 0) {
- recorder.recordValue(latencyMillis);
- cumulativeRecorder.recordValue(latencyMillis);
- }
-
- consumer.acknowledgeAsync(msg);
-
- if(arguments.poolMessages) {
- msg.release();
- }
- };
ClientBuilder clientBuilder = PulsarClient.builder() //
+ .enableTransaction(arguments.isEnableTransaction)
.serviceUrl(arguments.serviceURL) //
.connectionsPerBroker(arguments.maxConnections) //
.statsInterval(arguments.statsIntervalSeconds, TimeUnit.SECONDS) //
@@ -358,9 +360,121 @@ public class PerformanceConsumer {
if (isNotBlank(arguments.listenerName)) {
clientBuilder.listenerName(arguments.listenerName);
}
-
PulsarClient pulsarClient = clientBuilder.build();
+ AtomicReference<Transaction> atomicReference = buildTransaction(pulsarClient, arguments.isEnableTransaction,
+ arguments.transactionTimeout);
+
+ AtomicLong messageAckedCount = new AtomicLong();
+ Semaphore messageReceiveLimiter = new Semaphore(arguments.numMessagesPerTransaction);
+ Thread thread = Thread.currentThread();
+ MessageListener<ByteBuffer> listener = (consumer, msg) -> {
+ if (arguments.testTime > 0) {
+ if (System.nanoTime() > testEndTime) {
+ log.info("------------------- DONE -----------------------");
+ printAggregatedStats();
+ PerfClientUtils.exit(0);
+ thread.interrupt();
+ }
+ }
+ if (arguments.totalNumTxn > 0) {
+ if (totalEndTxnOpFailNum.sum() + totalEndTxnOpSuccessNum.sum() >= arguments.totalNumTxn) {
+ log.info("------------------- DONE -----------------------");
+ printAggregatedStats();
+ PerfClientUtils.exit(0);
+ thread.interrupt();
+ }
+ }
+ messagesReceived.increment();
+ bytesReceived.add(msg.size());
+
+ totalMessagesReceived.increment();
+ totalBytesReceived.add(msg.size());
+
+ if (limiter != null) {
+ limiter.acquire();
+ }
+
+ long latencyMillis = System.currentTimeMillis() - msg.getPublishTime();
+ if (latencyMillis >= 0) {
+ recorder.recordValue(latencyMillis);
+ cumulativeRecorder.recordValue(latencyMillis);
+ }
+ if (arguments.isEnableTransaction) {
+ try {
+ messageReceiveLimiter.acquire();
+ }catch (InterruptedException e){
+ log.error("Got error: ", e);
+ }
+ consumer.acknowledgeAsync(msg.getMessageId(), atomicReference.get()).thenRun(() -> {
+ totalMessageAck.increment();
+ messageAck.increment();
+ }).exceptionally(throwable ->{
+ log.error("Ack message {} failed with exception", msg, throwable);
+ totalMessageAckFailed.increment();
+ return null;
+ });
+ } else {
+ consumer.acknowledgeAsync(msg).thenRun(()->{
+ totalMessageAck.increment();
+ messageAck.increment();
+ }
+ ).exceptionally(throwable ->{
+ log.error("Ack message {} failed with exception", msg, throwable);
+ totalMessageAckFailed.increment();
+ return null;
+ }
+ );
+ }
+ if (arguments.poolMessages) {
+ msg.release();
+ }
+ if (arguments.isEnableTransaction
+ && messageAckedCount.incrementAndGet() == arguments.numMessagesPerTransaction) {
+ Transaction transaction = atomicReference.get();
+ if (!arguments.isAbortTransaction) {
+ transaction.commit()
+ .thenRun(() -> {
+ totalEndTxnOpSuccessNum.increment();
+ numTxnOpSuccess.increment();
+ })
+ .exceptionally(exception -> {
+ log.error("Commit transaction failed with exception : ", exception);
+ totalEndTxnOpFailNum.increment();
+ return null;
+ });
+ } else {
+ transaction.abort().thenRun(() -> {
+ log.info("Abort transaction {}", transaction.getTxnID().toString());
+ totalEndTxnOpSuccessNum.increment();
+ numTxnOpSuccess.increment();
+ }).exceptionally(exception -> {
+ log.error("Commit transaction {} failed with exception",
+ transaction.getTxnID().toString(),
+ exception);
+ totalEndTxnOpFailNum.increment();
+ return null;
+ });
+ }
+ while (true) {
+ try {
+ Transaction newTransaction = pulsarClient.newTransaction()
+ .withTransactionTimeout(arguments.transactionTimeout, TimeUnit.SECONDS)
+ .build().get();
+ atomicReference.compareAndSet(transaction, newTransaction);
+ totalNumTxnOpenSuccess.increment();
+ messageAckedCount.set(0);
+ messageReceiveLimiter.release(arguments.numMessagesPerTransaction);
+ break;
+ } catch (Exception e) {
+ log.error("Failed to new transaction with exception:", e);
+ totalNumTxnOpenFail.increment();
+ }
+ }
+ }
+
+ };
+
List<Future<Consumer<ByteBuffer>>> futures = Lists.newArrayList();
ConsumerBuilder<ByteBuffer> consumerBuilder = pulsarClient.newConsumer(Schema.BYTEBUFFER) //
.messageListener(listener) //
@@ -398,18 +512,16 @@ public class PerformanceConsumer {
}
}
}
-
for (Future<Consumer<ByteBuffer>> future : futures) {
future.get();
}
-
log.info("Start receiving from {} consumers per subscription on {} topics", arguments.numConsumers,
arguments.numTopics);
long start = System.nanoTime();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- printAggregatedThroughput(start);
+ printAggregatedThroughput(start, arguments);
printAggregatedStats();
}));
@@ -443,11 +555,27 @@ public class PerformanceConsumer {
long total = totalMessagesReceived.sum();
double rate = messagesReceived.sumThenReset() / elapsed;
double throughput = bytesReceived.sumThenReset() / elapsed * 8 / 1024 / 1024;
-
+ double rateAck = messageAck.sumThenReset() / elapsed;
+ long totalTxnOpSuccessNum = 0;
+ long totalTxnOpFailNum = 0;
+ double rateOpenTxn = 0;
reportHistogram = recorder.getIntervalHistogram(reportHistogram);
+ if (arguments.isEnableTransaction) {
+ totalTxnOpSuccessNum = totalEndTxnOpSuccessNum.sum();
+ totalTxnOpFailNum = totalEndTxnOpFailNum.sum();
+ rateOpenTxn = numTxnOpSuccess.sumThenReset() / elapsed;
+ log.info("--- Transaction: {} transaction end successfully --- {} transaction end failed "
+ + "--- {} Txn/s --- AckRate: {} msg/s",
+ totalTxnOpSuccessNum,
+ totalTxnOpFailNum,
+ dec.format(rateOpenTxn),
+ dec.format(rateAck));
+ }
log.info(
- "Throughput received: {} msg --- {} msg/s -- {} Mbit/s --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}",
+ "Throughput received: {} msg --- {} msg/s -- {} Mbit/s "
+ + "--- Latency: mean: {} ms - med: {} "
+ + "- 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}",
intFormat.format(total),
dec.format(rate), dec.format(throughput), dec.format(reportHistogram.getMean()),
reportHistogram.getValueAtPercentile(50), reportHistogram.getValueAtPercentile(95),
@@ -465,15 +593,41 @@ public class PerformanceConsumer {
pulsarClient.close();
}
- private static void printAggregatedThroughput(long start) {
+ private static void printAggregatedThroughput(long start, Arguments arguments) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesReceived.sum() / elapsed;
double throughput = totalBytesReceived.sum() / elapsed * 8 / 1024 / 1024;
+ long totalEndTxnSuccess = 0;
+ long totalEndTxnFail = 0;
+ long numTransactionOpenFailed = 0;
+ long numTransactionOpenSuccess = 0;
+ long totalnumMessageAckFailed = 0;
+ double rateAck = totalMessageAck.sum() / elapsed;
+ double rateOpenTxn = 0;
+ if (arguments.isEnableTransaction) {
+ totalEndTxnSuccess = totalEndTxnOpSuccessNum.sum();
+ totalEndTxnFail = totalEndTxnOpFailNum.sum();
+ rateOpenTxn = (totalEndTxnSuccess + totalEndTxnFail) / elapsed;
+ totalnumMessageAckFailed = totalMessageAckFailed.sum();
+ numTransactionOpenFailed = totalNumTxnOpenFail.sum();
+ numTransactionOpenSuccess = totalNumTxnOpenSuccess.sum();
+ log.info("-- Transaction: {} transaction end successfully --- {} transaction end failed "
+ + "--- {} transaction open successfully --- {} transaction open failed "
+ + "--- {} Txn/s ",
+ totalEndTxnSuccess,
+ totalEndTxnFail,
+ numTransactionOpenSuccess,
+ numTransactionOpenFailed,
+ dec.format(rateOpenTxn));
+ }
log.info(
- "Aggregated throughput stats --- {} records received --- {} msg/s --- {} Mbit/s",
- totalMessagesReceived,
+ "Aggregated throughput stats --- {} records received --- {} msg/s --- {} Mbit/s"
+ + "--- AckRate: {} msg/s --- ack failed {} msg",
+ totalMessagesReceived.sum(),
dec.format(rate),
- dec.format(throughput));
+ dec.format(throughput),
+ rateAck,
+ totalnumMessageAckFailed);
}
private static void printAggregatedStats() {
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
index fee96a8..cfd5568 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -49,7 +49,11 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import org.HdrHistogram.Histogram;
@@ -70,7 +74,9 @@ import org.apache.pulsar.client.api.TypedMessageBuilder;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_MAX_PENDING_MESSAGES_ACROSS_PARTITIONS;
import static org.apache.pulsar.client.impl.conf.ProducerConfigurationData.DEFAULT_BATCHING_MAX_MESSAGES;
+import static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
import org.slf4j.Logger;
@@ -88,11 +94,18 @@ public class PerformanceProducer {
private static final LongAdder messagesFailed = new LongAdder();
private static final LongAdder bytesSent = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
private static final LongAdder totalMessagesSent = new LongAdder();
private static final LongAdder totalBytesSent = new LongAdder();
- private static Recorder recorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
- private static Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder recorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder cumulativeRecorder = new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+ private static final LongAdder totalEndTxnOpSuccessNum = new LongAdder();
+ private static final LongAdder totalEndTxnOpFailNum = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
private static IMessageFormatter messageFormatter = null;
@@ -238,7 +251,7 @@ public class PerformanceProducer {
public String messageKeyGenerationMode = null;
@Parameter(names = {"-ioThreads", "--num-io-threads"}, description = "Set the number of threads to be " +
- "used for handling connections to brokers, default is 1 thread")
+ "used for handling connections to brokers. The default value is 1.")
public int ioThreads = 1;
@Parameter(names = {"-bw", "--busy-wait"}, description = "Enable Busy-Wait on the Pulsar client")
@@ -254,6 +267,22 @@ public class PerformanceProducer {
@Parameter(names = {"-fc", "--format-class"}, description="Custom Formatter class name")
public String formatterClass = "org.apache.pulsar.testclient.DefaultMessageFormatter";
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 10;
+
+ @Parameter(names = {"-nmt", "--numMessage-perTransaction"},
+ description = "The number of messages sent by a transaction. "
+ + "(After --txn-enable setting to true, -nmt takes effect)")
+ public int numMessagesPerTransaction = 50;
+
+ @Parameter(names = {"-txn", "--txn-enable"}, description = "Enable or disable the transaction")
+ public boolean isEnableTransaction = false;
+
+ @Parameter(names = {"-abort"}, description = "Abort the transaction. (After --txn-enable "
+ + "setting to true, -abort takes effect)")
+ public boolean isAbortTransaction = false;
+
@Parameter(names = { "--histogram-file" }, description = "HdrHistogram output file")
public String histogramFile = null;
}
@@ -376,7 +405,7 @@ public class PerformanceProducer {
long start = System.nanoTime();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
- printAggregatedThroughput(start);
+ printAggregatedThroughput(start, arguments);
printAggregatedStats();
}));
@@ -467,14 +496,27 @@ public class PerformanceProducer {
long now = System.nanoTime();
double elapsed = (now - oldTime) / 1e9;
long total = totalMessagesSent.sum();
+ long totalTxnOpSuccess = 0;
+ long totalTxnOpFail = 0;
+ double rateOpenTxn = 0;
double rate = messagesSent.sumThenReset() / elapsed;
double failureRate = messagesFailed.sumThenReset() / elapsed;
double throughput = bytesSent.sumThenReset() / elapsed / 1024 / 1024 * 8;
reportHistogram = recorder.getIntervalHistogram(reportHistogram);
+ if (arguments.isEnableTransaction) {
+ totalTxnOpSuccess = totalEndTxnOpSuccessNum.sum();
+ totalTxnOpFail = totalEndTxnOpFailNum.sum();
+ rateOpenTxn = numTxnOpSuccess.sumThenReset() / elapsed;
+ log.info("--- Transaction : {} transaction end successfully ---{} transaction end failed "
+ + "--- {} Txn/s",
+ totalTxnOpSuccess, totalTxnOpFail, totalFormat.format(rateOpenTxn));
+ }
log.info(
- "Throughput produced: {} msg --- {} msg/s --- {} Mbit/s --- failure {} msg/s --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}",
+ "Throughput produced: {} msg --- {} msg/s --- {} Mbit/s --- failure {} msg/s "
+ + "--- Latency: mean: "
+ + "{} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}",
intFormat.format(total),
throughputFormat.format(rate), throughputFormat.format(throughput),
throughputFormat.format(failureRate),
@@ -520,6 +562,7 @@ public class PerformanceProducer {
List<Future<Producer<byte[]>>> futures = Lists.newArrayList();
ClientBuilder clientBuilder = PulsarClient.builder() //
+ .enableTransaction(arguments.isEnableTransaction)//
.serviceUrl(arguments.serviceURL) //
.connectionsPerBroker(arguments.maxConnections) //
.ioThreads(arguments.ioThreads) //
@@ -549,6 +592,9 @@ public class PerformanceProducer {
// enable round robin message routing if it is a partitioned topic
.messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
+ if (arguments.isEnableTransaction) {
+ producerBuilder.sendTimeout(0, TimeUnit.SECONDS);
+ }
if (arguments.producerName != null) {
String producerName = String.format("%s%s%d", arguments.producerName, arguments.separator, producerId);
producerBuilder.producerName(producerName);
@@ -613,6 +659,10 @@ public class PerformanceProducer {
}
// Send messages on all topics/producers
long totalSent = 0;
+ AtomicReference<Transaction> transactionAtomicReference = buildTransaction(client,
+ arguments.isEnableTransaction, arguments.transactionTimeout);
+ AtomicLong numMessageSend = new AtomicLong(0);
+ Semaphore numMsgPerTxnLimit = new Semaphore(arguments.numMessagesPerTransaction);
while (true) {
for (Producer<byte[]> producer : producers) {
if (arguments.testTime > 0) {
@@ -635,7 +685,8 @@ public class PerformanceProducer {
}
}
rateLimiter.acquire();
-
+ //if transaction is disable, transaction will be null.
+ Transaction transaction = transactionAtomicReference.get();
final long sendTime = System.nanoTime();
byte[] payloadData;
@@ -650,10 +701,22 @@ public class PerformanceProducer {
} else {
payloadData = payloadBytes;
}
-
- TypedMessageBuilder<byte[]> messageBuilder = producer.newMessage()
- .value(payloadData);
- if (arguments.delay >0) {
+ TypedMessageBuilder<byte[]> messageBuilder;
+ if (arguments.isEnableTransaction) {
+ if (arguments.numMessagesPerTransaction> 0) {
+ try{
+ numMsgPerTxnLimit.acquire();
+ }catch (InterruptedException exception){
+ log.error("Get exception: ", exception);
+ }
+ }
+ messageBuilder = producer.newMessage(transaction)
+ .value(payloadData);
+ } else {
+ messageBuilder = producer.newMessage()
+ .value(payloadData);
+ }
+ if (arguments.delay > 0) {
messageBuilder.deliverAfter(arguments.delay, TimeUnit.SECONDS);
}
//generate msg key
@@ -662,8 +725,8 @@ public class PerformanceProducer {
} else if (msgKeyMode == MessageKeyGenerationMode.autoIncrement) {
messageBuilder.key(String.valueOf(totalSent));
}
+ PulsarClient pulsarClient = client;
messageBuilder.sendAsync().thenRun(() -> {
- messagesSent.increment();
bytesSent.add(payloadData.length);
totalMessagesSent.increment();
@@ -681,13 +744,58 @@ public class PerformanceProducer {
if (ex.getCause() instanceof ArrayIndexOutOfBoundsException) {
return null;
}
- log.warn("Write error on message", ex);
+ log.warn("Write message error with exception", ex);
messagesFailed.increment();
if (arguments.exitOnFailure) {
PerfClientUtils.exit(-1);
}
return null;
});
+ if (arguments.isEnableTransaction
+ && numMessageSend.incrementAndGet() == arguments.numMessagesPerTransaction) {
+ if (!arguments.isAbortTransaction) {
+ transaction.commit()
+ .thenRun(() -> {
+ log.info("Committed transaction {}",
+ transaction.getTxnID().toString());
+ totalEndTxnOpSuccessNum.increment();
+ numTxnOpSuccess.increment();
+ })
+ .exceptionally(exception -> {
+ log.error("Commit transaction failed with exception : ",
+ exception);
+ totalEndTxnOpFailNum.increment();
+ return null;
+ });
+ } else {
+ transaction.abort().thenRun(() -> {
+ log.info("Abort transaction {}", transaction.getTxnID().toString());
+ totalEndTxnOpSuccessNum.increment();
+ numTxnOpSuccess.increment();
+ }).exceptionally(exception -> {
+ log.error("Commit transaction {} failed with exception",
+ transaction.getTxnID().toString(),
+ exception);
+ totalEndTxnOpFailNum.increment();
+ return null;
+ });
+ }
+ while(true) {
+ try {
+ Transaction newTransaction = pulsarClient.newTransaction()
+ .withTransactionTimeout(arguments.transactionTimeout,
+ TimeUnit.SECONDS).build().get();
+ transactionAtomicReference.compareAndSet(transaction, newTransaction);
+ numMessageSend.set(0);
+ numMsgPerTxnLimit.release(arguments.numMessagesPerTransaction);
+ totalNumTxnOpenTxnSuccess.increment();
+ break;
+ }catch (Exception e){
+ totalNumTxnOpenTxnFail.increment();
+ log.error("Failed to new transaction with exception: ", e);
+ }
+ }
+ }
}
}
} catch (Throwable t) {
@@ -696,6 +804,7 @@ public class PerformanceProducer {
if (null != client) {
try {
client.close();
+ PerfClientUtils.exit(-1);
} catch (PulsarClientException e) {
log.error("Failed to close test client", e);
}
@@ -703,13 +812,34 @@ public class PerformanceProducer {
}
}
- private static void printAggregatedThroughput(long start) {
+ private static void printAggregatedThroughput(long start, Arguments arguments) {
double elapsed = (System.nanoTime() - start) / 1e9;
double rate = totalMessagesSent.sum() / elapsed;
double throughput = totalBytesSent.sum() / elapsed / 1024 / 1024 * 8;
+ long totalTxnSuccess = 0;
+ long totalTxnFail = 0;
+ double rateOpenTxn = 0;
+ long numTransactionOpenFailed = 0;
+ long numTransactionOpenSuccess = 0;
+
+ if (arguments.isEnableTransaction) {
+ totalTxnSuccess = totalEndTxnOpSuccessNum.sum();
+ totalTxnFail = totalEndTxnOpFailNum.sum();
+ rateOpenTxn = elapsed / (totalTxnFail + totalTxnSuccess);
+ numTransactionOpenFailed = totalNumTxnOpenTxnFail.sum();
+ numTransactionOpenSuccess = totalNumTxnOpenTxnSuccess.sum();
+ log.info("--- Transaction : {} transaction end successfully --- {} transaction end failed "
+ + "--- {} transaction open successfully --- {} transaction open failed "
+ + "--- {} Txn/s",
+ totalTxnSuccess,
+ totalTxnFail,
+ numTransactionOpenSuccess,
+ numTransactionOpenFailed,
+ totalFormat.format(rateOpenTxn));
+ }
log.info(
- "Aggregated throughput stats --- {} records sent --- {} msg/s --- {} Mbit/s",
- totalMessagesSent,
+ "Aggregated throughput stats --- {} records sent --- {} msg/s --- {} Mbit/s ",
+ totalMessagesSent.sum(),
totalFormat.format(rate),
totalFormat.format(throughput));
}
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
new file mode 100644
index 0000000..49d441e
--- /dev/null
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceTransaction.java
@@ -0,0 +1,707 @@
+/**
+ * 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.testclient;
+
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static org.apache.pulsar.testclient.utils.PerformanceUtils.buildTransaction;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.google.common.collect.Lists;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
+import org.HdrHistogram.Histogram;
+import org.HdrHistogram.HistogramLogWriter;
+import org.HdrHistogram.Recorder;
+import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+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.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import org.apache.pulsar.testclient.utils.PaddingDecimalFormat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceTransaction {
+
+
+ private static final LongAdder totalNumEndTxnOpFailed = new LongAdder();
+ private static final LongAdder totalNumEndTxnOpSuccess = new LongAdder();
+ private static final LongAdder numTxnOpSuccess = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnFail = new LongAdder();
+ private static final LongAdder totalNumTxnOpenTxnSuccess = new LongAdder();
+
+ private static final LongAdder numMessagesAckFailed = new LongAdder();
+ private static final LongAdder numMessagesAckSuccess = new LongAdder();
+ private static final LongAdder numMessagesSendFailed = new LongAdder();
+ private static final LongAdder numMessagesSendSuccess = new LongAdder();
+
+ private static final Recorder messageAckRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageAckCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+ private static final Recorder messageSendRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+ private static final Recorder messageSendRCumulativeRecorder =
+ new Recorder(TimeUnit.SECONDS.toMicros(120000), 5);
+
+
+ @Parameters(commandDescription = "Test pulsar transaction performance.")
+ static class Arguments {
+
+ @Parameter(names = {"-h", "--help"}, description = "Help message", help = true)
+ boolean help;
+
+ @Parameter(names = {"--conf-file"}, description = "Configuration file")
+ public String confFile;
+
+ @Parameter(names = "--topics-c", description = "All topics that need ack for a transaction", required =
+ true)
+ public List<String> consumerTopic = Collections.singletonList("test-consume");
+
+ @Parameter(names = "--topics-p", description = "All topics that need produce for a transaction",
+ required = true)
+ public List<String> producerTopic = Collections.singletonList("test-produce");
+
+ @Parameter(names = {"-threads", "--num-test-threads"}, description = "Number of test threads."
+ + "This thread is for a new transaction to ack messages from consumer topics and produce message to "
+ + "producer topics, and then commit or abort this transaction. "
+ + "Increasing the number of threads increases the parallelism of the performance test, "
+ + "thereby increasing the intensity of the stress test.")
+ public int numTestThreads = 1;
+
+ @Parameter(names = {"-au", "--admin-url"}, description = "Pulsar Admin URL")
+ public String adminURL;
+
+ @Parameter(names = {"-u", "--service-url"}, description = "Pulsar Service URL")
+ public String serviceURL;
+
+ @Parameter(names = {"-np",
+ "--partitions"}, description = "Create partitioned topics with a given number of partitions, 0 means"
+ + "not trying to create a topic")
+ public Integer partitions = null;
+
+ @Parameter(names = {"-c",
+ "--max-connections"}, description = "Max number of TCP connections to a single broker")
+ public int maxConnections = 100;
+
+ @Parameter(names = {"-time",
+ "--test-duration"}, description = "Test duration (in second). 0 means keeping publishing")
+ public long testTime = 0;
+
+ @Parameter(names = {"-ioThreads", "--num-io-threads"}, description = "Set the number of threads to be " +
+ "used for handling connections to brokers. The default value is 1.")
+ public int ioThreads = 1;
+
+ @Parameter(names = {"-ss",
+ "--subscriptions"}, description = "A list of subscriptions to consume (for example, sub1,sub2)")
+ public List<String> subscriptions = Collections.singletonList("sub");
+
+ @Parameter(names = {"-ns", "--num-subscriptions"}, description = "Number of subscriptions (per topic)")
+ public int numSubscriptions = 1;
+
+ @Parameter(names = {"-sp", "--subscription-position"}, description = "Subscription position")
+ private SubscriptionInitialPosition subscriptionInitialPosition = SubscriptionInitialPosition.Earliest;
+
+ @Parameter(names = {"-st", "--subscription-type"}, description = "Subscription type")
+ public SubscriptionType subscriptionType = SubscriptionType.Shared;
+
+ @Parameter(names = {"-q", "--receiver-queue-size"}, description = "Size of the receiver queue")
+ public int receiverQueueSize = 1000;
+
+ @Parameter(names = {"-tto", "--txn-timeout"}, description = "Set the time value of transaction timeout,"
+ + " and the time unit is second. (After --txn-enable setting to true, --txn-timeout takes effect)")
+ public long transactionTimeout = 5;
+
+ @Parameter(names = {"-ntxn",
+ "--number-txn"}, description = "Set the number of transaction. 0 means keeping open."
+ + "If transaction disabled, it means the number of tasks. The task or transaction produces or "
+ + "consumes a specified number of messages.")
+ public long numTransactions = 0;
+
+ @Parameter(names = {"-nmp", "--numMessage-perTransaction-produce"},
+ description = "Set the number of messages produced in a transaction."
+ + "If transaction disabled, it means the number of messages produced in a task.")
+ public int numMessagesProducedPerTransaction = 1;
+
+ @Parameter(names = {"-nmc", "--numMessage-perTransaction-consume"},
+ description = "Set the number of messages consumed in a transaction."
+ + "If transaction disabled, it means the number of messages consumed in a task.")
+ public int numMessagesReceivedPerTransaction = 1;
+
+ @Parameter(names = {"--txn-disable"}, description = "Disable transaction")
+ public boolean isDisableTransaction = false;
+
+ @Parameter(names = {"-abort"}, description = "Abort the transaction. (After --txn-disEnable "
+ + "setting to false, -abort takes effect)")
+ public boolean isAbortTransaction = false;
+
+ @Parameter(names = "-txnRate", description = "Set the rate of opened transaction or task. 0 means no limit")
+ public int openTxnRate = 0;
+ }
+
+ public static void main(String[] args)
+ throws IOException, PulsarAdminException, ExecutionException, InterruptedException {
+ final Arguments arguments = new Arguments();
+ JCommander jc = new JCommander(arguments);
+ jc.setProgramName("pulsar-perf transaction");
+
+ try {
+ jc.parse(args);
+ } catch (ParameterException e) {
+ System.out.println(e.getMessage());
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+ if (arguments.help) {
+ jc.usage();
+ PerfClientUtils.exit(-1);
+ }
+
+
+ if (arguments.confFile != null) {
+ Properties prop = new Properties(System.getProperties());
+ prop.load(new FileInputStream(arguments.confFile));
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("brokerServiceUrl");
+ }
+
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("webServiceUrl");
+ }
+
+ // fallback to previous-version serviceUrl property to maintain backward-compatibility
+ if (arguments.serviceURL == null) {
+ arguments.serviceURL = prop.getProperty("serviceUrl", "http://localhost:8080/");
+ }
+
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("webServiceUrl");
+ }
+ if (arguments.adminURL == null) {
+ arguments.adminURL = prop.getProperty("adminURL", "http://localhost:8080/");
+ }
+ }
+
+
+ // Dump config variables
+ PerfClientUtils.printJVMInformation(log);
+
+ ObjectMapper m = new ObjectMapper();
+ ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+ log.info("Starting Pulsar perf transaction with config: {}", w.writeValueAsString(arguments));
+
+ final byte[] payloadBytes = new byte[1024];
+ Random random = new Random(0);
+ for (int i = 0; i < payloadBytes.length; ++i) {
+ payloadBytes[i] = (byte) (random.nextInt(26) + 65);
+ }
+ if (arguments.partitions != null) {
+ PulsarAdminBuilder clientBuilder = PulsarAdmin.builder()
+ .serviceHttpUrl(arguments.adminURL);
+ try (PulsarAdmin client = clientBuilder.build()) {
+ for (String topic : arguments.producerTopic) {
+ log.info("Creating produce partitioned topic {} with {} partitions", topic, arguments.partitions);
+ try {
+ client.topics().createPartitionedTopic(topic, arguments.partitions);
+ } catch (PulsarAdminException.ConflictException alreadyExists) {
+ if (log.isDebugEnabled()) {
+ log.debug("Topic {} already exists: {}", topic, alreadyExists);
+ }
+ PartitionedTopicMetadata partitionedTopicMetadata =
+ client.topics().getPartitionedTopicMetadata(topic);
+ if (partitionedTopicMetadata.partitions != arguments.partitions) {
+ log.error(
+ "Topic {} already exists but it has a wrong number of partitions: {}, expecting {}",
+ topic, partitionedTopicMetadata.partitions, arguments.partitions);
+ PerfClientUtils.exit(-1);
+ }
+ }
+ }
+ }
+ }
+
+ PulsarClient client =
+ PulsarClient.builder().enableTransaction(!arguments.isDisableTransaction)
+ .serviceUrl(arguments.serviceURL)
+ .connectionsPerBroker(arguments.maxConnections)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .ioThreads(arguments.ioThreads)
+ .build();
+
+ ExecutorService executorService = new ThreadPoolExecutor(arguments.numTestThreads,
+ arguments.numTestThreads,
+ 0L, TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<Runnable>());
+
+
+ long startTime = System.nanoTime();
+ long testEndTime = startTime + (long) (arguments.testTime * 1e9);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ if (!arguments.isDisableTransaction) {
+ printTxnAggregatedThroughput(startTime);
+ } else {
+ printAggregatedThroughput(startTime);
+ }
+ printAggregatedStats();
+ }));
+
+ // start perf test
+ AtomicBoolean executing = new AtomicBoolean(true);
+
+ RateLimiter rateLimiter = arguments.openTxnRate > 0
+ ? RateLimiter.create(arguments.openTxnRate)
+ : null;
+ for(int i = 0; i < arguments.numTestThreads; i++) {
+ executorService.submit(() -> {
+ //The producer and consumer clients are built in advance, and then this thread is
+ //responsible for the production and consumption tasks of the transaction through the loop.
+ //A thread may perform tasks of multiple transactions in a traversing manner.
+ List<Producer<byte[]>> producers = null;
+ List<List<Consumer<byte[]>>> consumers = null;
+ try {
+ producers = buildProducers(client, arguments);
+ consumers = buildConsumer(client, arguments);
+ } catch (Exception e) {
+ log.error("Failed to build Producer/Consumer with exception : ", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ AtomicReference<Transaction> atomicReference = buildTransaction(client,
+ !arguments.isDisableTransaction, arguments.transactionTimeout);
+ //The while loop has no break, and finally ends the execution through the shutdownNow of
+ //the executorService
+ while (true) {
+ if (arguments.numTransactions > 0) {
+ if (totalNumTxnOpenTxnFail.sum()
+ + totalNumTxnOpenTxnSuccess.sum() >= arguments.numTransactions) {
+ if (totalNumEndTxnOpFailed.sum()
+ + totalNumEndTxnOpSuccess.sum() < arguments.numTransactions ) {
+ continue;
+ }
+ log.info("------------------- DONE -----------------------");
+ executing.compareAndSet(true, false);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(0);
+ break;
+ }
+ }
+ if (arguments.testTime > 0) {
+ if (System.nanoTime() > testEndTime) {
+ log.info("------------------- DONE -----------------------");
+ executing.compareAndSet(true, false);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(0);
+ break;
+ }
+ }
+ Transaction transaction = atomicReference.get();
+ for (List<Consumer<byte[]>> subscriptions : consumers) {
+ for (Consumer<byte[]> consumer : subscriptions) {
+ for (int j = 0; j < arguments.numMessagesReceivedPerTransaction; j++) {
+ Message message = null;
+ try {
+ message = consumer.receive();
+ } catch (PulsarClientException e) {
+ log.error("Receive message failed", e);
+ executorService.shutdownNow();
+ PerfClientUtils.exit(-1);
+ }
+ long receiveTime = System.nanoTime();
+ if (!arguments.isDisableTransaction) {
+ consumer.acknowledgeAsync(message.getMessageId(), transaction)
+ .thenRun(() -> {
+ long latencyMicros = NANOSECONDS.toMicros(
+ System.nanoTime() - receiveTime);
+ messageAckRecorder.recordValue(latencyMicros);
+ messageAckCumulativeRecorder.recordValue(latencyMicros);
+ numMessagesAckSuccess.increment();
+ }).exceptionally(exception -> {
+ if (exception instanceof InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message failed with transaction {} throw exception",
+ transaction, exception);
+ numMessagesAckFailed.increment();
+ return null;
+ });
+ } else {
+ consumer.acknowledgeAsync(message).thenRun(() -> {
+ long latencyMicros = NANOSECONDS.toMicros(
+ System.nanoTime() - receiveTime);
+ messageAckRecorder.recordValue(latencyMicros);
+ messageAckCumulativeRecorder.recordValue(latencyMicros);
+ numMessagesAckSuccess.increment();
+ }).exceptionally(exception -> {
+ if (exception instanceof InterruptedException && !executing.get()) {
+ return null;
+ }
+ log.error(
+ "Ack message failed with transaction {} throw exception",
+ transaction, exception);
+ numMessagesAckFailed.increment();
+ return null;
+ });
+ }
+ }
+ }
+ }
+
+ for(Producer<byte[]> producer : producers){
+ for (int j = 0; j < arguments.numMessagesProducedPerTransaction; j++) {
+ long sendTime = System.nanoTime();
+ if (!arguments.isDisableTransaction) {
+ producer.newMessage(transaction).value(payloadBytes)
+ .sendAsync().thenRun(() -> {
+ long latencyMicros = NANOSECONDS.toMicros(
+ System.nanoTime() - sendTime);
+ messageSendRecorder.recordValue(latencyMicros);
+ messageSendRCumulativeRecorder.recordValue(latencyMicros);
+ numMessagesSendSuccess.increment();
+ }).exceptionally(exception -> {
+ if (exception instanceof InterruptedException && ! executing.get()) {
+ return null;
+ }
+ log.error("Send transaction message failed with exception : ", exception);
+ numMessagesSendFailed.increment();
+ return null;
+ });
+ } else {
+ producer.newMessage().value(payloadBytes)
+ .sendAsync().thenRun(() -> {
+ long latencyMicros = NANOSECONDS.toMicros(
+ System.nanoTime() - sendTime);
+ messageSendRecorder.recordValue(latencyMicros);
+ messageSendRCumulativeRecorder.recordValue(latencyMicros);
+ numMessagesSendSuccess.increment();
+ }).exceptionally(exception -> {
+ if (exception instanceof InterruptedException && ! executing.get()) {
+ return null;
+ }
+ log.error("Send message failed with exception : ", exception);
+ numMessagesSendFailed.increment();
+ return null;
+ });
+ }
+ }
+ }
+
+ if (rateLimiter != null){
+ rateLimiter.tryAcquire();
+ }
+ if (!arguments.isDisableTransaction) {
+ if (!arguments.isAbortTransaction) {
+ transaction.commit()
+ .thenRun(() -> {
+ numTxnOpSuccess.increment();
+ totalNumEndTxnOpSuccess.increment();
+ }).exceptionally(exception -> {
+ if (exception instanceof InterruptedException && ! executing.get()) {
+ return null;
+ }
+ log.error("Commit transaction {} failed with exception",
+ transaction.getTxnID().toString(),
+ exception);
+ totalNumEndTxnOpFailed.increment();
+ return null;
+ });
+ } else {
+ transaction.abort().thenRun(() -> {
+ numTxnOpSuccess.increment();
+ totalNumEndTxnOpSuccess.increment();
+ }).exceptionally(exception -> {
+ if (exception instanceof InterruptedException && ! executing.get()) {
+ return null;
+ }
+ log.error("Commit transaction {} failed with exception",
+ transaction.getTxnID().toString(),
+ exception);
+ totalNumEndTxnOpFailed.increment();
+ return null;
+ });
+ }
+ while (true) {
+ try{
+ Transaction newTransaction = client.newTransaction()
+ .withTransactionTimeout(arguments.transactionTimeout, TimeUnit.SECONDS)
+ .build()
+ .get();
+ atomicReference.compareAndSet(transaction, newTransaction);
+ totalNumTxnOpenTxnSuccess.increment();
+ break;
+ }catch (Exception throwable){
+ if (throwable instanceof InterruptedException && !executing.get()) {
+ break;
+ }
+ log.error("Failed to new transaction with exception: ", throwable);
+ totalNumTxnOpenTxnFail.increment();
+ }
+ }
+ } else {
+ totalNumTxnOpenTxnSuccess.increment();
+ totalNumEndTxnOpSuccess.increment();
+ numTxnOpSuccess.increment();
+ }
+ }
+ });
+ }
+
+
+
+ // Print report stats
+ long oldTime = System.nanoTime();
+
+ Histogram reportSendHistogram = null;
+ Histogram reportAckHistogram = null;
+
+ String statsFileName = "perf-transaction-" + System.currentTimeMillis() + ".hgrm";
+ log.info("Dumping latency stats to {}", statsFileName);
+
+ PrintStream histogramLog = new PrintStream(new FileOutputStream(statsFileName), false);
+ HistogramLogWriter histogramLogWriter = new HistogramLogWriter(histogramLog);
+
+ // Some log header bits
+ histogramLogWriter.outputLogFormatVersion();
+ histogramLogWriter.outputLegend();
+
+ while (executing.get()) {
+ try {
+ Thread.sleep(10000);
+ } catch (InterruptedException e) {
+ break;
+ }
+ long now = System.nanoTime();
+ double elapsed = (now - oldTime) / 1e9;
+ long total = totalNumEndTxnOpFailed.sum() + totalNumTxnOpenTxnSuccess.sum();
+ double rate = numTxnOpSuccess.sumThenReset() / elapsed;
+ reportSendHistogram = messageSendRecorder.getIntervalHistogram(reportSendHistogram);
+ reportAckHistogram = messageAckRecorder.getIntervalHistogram(reportAckHistogram);
+ String txnOrTaskLog = !arguments.isDisableTransaction
+ ? "Throughput transaction: {} transaction executes --- {} transaction/s"
+ : "Throughput task: {} task executes --- {} task/s";
+ log.info(
+ txnOrTaskLog + " ---send Latency: mean: {} ms - med: {} "
+ + "- 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}" + "---ack Latency: "
+ + "mean: {} ms - med: {} - 95pct: {} - 99pct: {} - 99.9pct: {} - 99.99pct: {} - Max: {}",
+ intFormat.format(total),
+ dec.format(rate),
+ dec.format(reportSendHistogram.getMean() / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(50) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(95) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99.9) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99.99) / 1000.0),
+ dec.format(reportSendHistogram.getMaxValue() / 1000.0),
+ dec.format(reportAckHistogram.getMean() / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(50) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(95) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99.9) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99.99) / 1000.0),
+ dec.format(reportAckHistogram.getMaxValue() / 1000.0));
+
+ histogramLogWriter.outputIntervalHistogram(reportSendHistogram);
+ histogramLogWriter.outputIntervalHistogram(reportAckHistogram);
+ reportSendHistogram.reset();
+ reportAckHistogram.reset();
+
+ oldTime = now;
+ }
+
+
+ }
+
+
+ private static void printTxnAggregatedThroughput(long start) {
+ double elapsed = (System.nanoTime() - start) / 1e9;
+ long numTransactionEndFailed = totalNumEndTxnOpFailed.sum();
+ long numTransactionEndSuccess = totalNumEndTxnOpSuccess.sum();
+ long total = numTransactionEndFailed + numTransactionEndSuccess;
+ double rate = total / elapsed;
+ long numMessageAckFailed = numMessagesAckFailed.sum();
+ long numMessageAckSuccess = numMessagesAckSuccess.sum();
+ long numMessageSendFailed = numMessagesSendFailed.sum();
+ long numMessageSendSuccess = numMessagesSendSuccess.sum();
+ long numTransactionOpenFailed = totalNumTxnOpenTxnFail.sum();
+ long numTransactionOpenSuccess = totalNumTxnOpenTxnSuccess.sum();
+
+ log.info(
+ "Aggregated throughput stats --- {} transaction executed --- {} transaction/s "
+ + " --- {} transaction open successfully --- {} transaction open failed"
+ + " --- {} transaction end successfully --- {} transaction end failed"
+ + "--- {} message ack failed --- {} message send failed"
+ + "--- {} message ack success --- {} message send success ",
+ total,
+ dec.format(rate),
+ numTransactionOpenSuccess,
+ numTransactionOpenFailed,
+ numTransactionEndSuccess,
+ numTransactionEndFailed,
+ numMessageAckFailed,
+ numMessageSendFailed,
+ numMessageAckSuccess,
+ numMessageSendSuccess);
+
+ }
+
+ private static void printAggregatedThroughput(long start) {
+ double elapsed = (System.nanoTime() - start) / 1e9;
+ long total = totalNumEndTxnOpFailed.sum() + totalNumEndTxnOpSuccess.sum();
+ double rate = total / elapsed;
+ long numMessageAckFailed = numMessagesAckFailed.sum();
+ long numMessageAckSuccess = numMessagesAckSuccess.sum();
+ long numMessageSendFailed = numMessagesSendFailed.sum();
+ long numMessageSendSuccess = numMessagesSendSuccess.sum();
+ log.info(
+ "Aggregated throughput stats --- {} task executed --- {} task/s "
+ + "--- {} message ack failed --- {} message send failed"
+ + "--- {} message ack success --- {} message send success ",
+ total,
+ totalFormat.format(rate),
+ numMessageAckFailed,
+ numMessageSendFailed,
+ numMessageAckSuccess,
+ numMessageSendSuccess);
+ }
+
+ private static void printAggregatedStats() {
+ Histogram reportAckHistogram = messageAckCumulativeRecorder.getIntervalHistogram();
+ Histogram reportSendHistogram = messageSendRCumulativeRecorder.getIntervalHistogram();
+ log.info(
+ "Messages ack aggregated latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {} - "
+ + "99.9pct: {} - "
+ + "99.99pct: {} - 99.999pct: {} - Max: {}",
+ dec.format(reportAckHistogram.getMean() / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(50) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(95) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99.9) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99.99) / 1000.0),
+ dec.format(reportAckHistogram.getValueAtPercentile(99.999) / 1000.0),
+ dec.format(reportAckHistogram.getMaxValue() / 1000.0));
+ log.info(
+ "Messages send aggregated latency stats --- Latency: mean: {} ms - med: {} - 95pct: {} - 99pct: {} - "
+ + "99.9pct: {} - "
+ + "99.99pct: {} - 99.999pct: {} - Max: {}",
+ dec.format(reportSendHistogram.getMean() / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(50) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(95) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99.9) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99.99) / 1000.0),
+ dec.format(reportSendHistogram.getValueAtPercentile(99.999) / 1000.0),
+ dec.format(reportSendHistogram.getMaxValue() / 1000.0));
+ }
+
+
+
+ static final DecimalFormat dec = new PaddingDecimalFormat("0.000", 7);
+ static final DecimalFormat intFormat = new PaddingDecimalFormat("0", 7);
+ static final DecimalFormat totalFormat = new DecimalFormat("0.000");
+ private static final Logger log = LoggerFactory.getLogger(PerformanceProducer.class);
+
+
+ private static List<List<Consumer<byte[]>>> buildConsumer(PulsarClient client, Arguments arguments)
+ throws ExecutionException, InterruptedException {
+ ConsumerBuilder<byte[]> consumerBuilder = client.newConsumer(Schema.BYTES) //
+ .subscriptionType(arguments.subscriptionType)
+ .receiverQueueSize(arguments.receiverQueueSize)
+ .subscriptionInitialPosition(arguments.subscriptionInitialPosition);
+
+ Iterator<String> consumerTopicsIterator = arguments.consumerTopic.iterator();
+ List<List<Consumer<byte[]>>> consumers = Lists.newArrayListWithCapacity(arguments.consumerTopic.size());
+ while(consumerTopicsIterator.hasNext()){
+ String topic = consumerTopicsIterator.next();
+ final List<Consumer<byte[]>> subscriptions = Lists.newArrayListWithCapacity(arguments.numSubscriptions);
+ final List<Future<Consumer<byte[]>>> subscriptionFutures =
+ Lists.newArrayListWithCapacity(arguments.numSubscriptions);
+ log.info("Create subscriptions for topic {}", topic);
+ for (int j = 0; j < arguments.numSubscriptions; j++) {
+ String subscriberName = arguments.subscriptions.get(j);
+ subscriptionFutures
+ .add(consumerBuilder.clone().topic(topic).subscriptionName(subscriberName)
+ .subscribeAsync());
+ }
+ for (Future<Consumer<byte[]>> future : subscriptionFutures) {
+ subscriptions.add(future.get());
+ }
+ consumers.add(subscriptions);
+ }
+ return consumers;
+ }
+
+ private static List<Producer<byte[]>> buildProducers(PulsarClient client, Arguments arguments)
+ throws ExecutionException, InterruptedException {
+
+ ProducerBuilder<byte[]> producerBuilder = client.newProducer(Schema.BYTES)
+ .sendTimeout(0, TimeUnit.SECONDS);
+
+ final List<Future<Producer<byte[]>>> producerFutures = Lists.newArrayList();
+ Iterator<String> produceTopicsIterator = arguments.producerTopic.iterator();
+ while(produceTopicsIterator.hasNext()){
+ String topic = produceTopicsIterator.next();
+ log.info("Create producer for topic {}", topic);
+ producerFutures.add(producerBuilder.clone().topic(topic).createAsync());
+ }
+ final List<Producer<byte[]>> producers = Lists.newArrayListWithCapacity(producerFutures.size());
+
+ for (Future<Producer<byte[]>> future : producerFutures) {
+ producers.add(future.get());
+ }
+ return producers;
+ }
+
+}
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/utils/PerformanceUtils.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/utils/PerformanceUtils.java
new file mode 100644
index 0000000..ded1131
--- /dev/null
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/utils/PerformanceUtils.java
@@ -0,0 +1,59 @@
+/**
+ * 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.testclient.utils;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.transaction.Transaction;
+import org.apache.pulsar.testclient.PerformanceProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PerformanceUtils {
+
+ private static final Logger log = LoggerFactory.getLogger(PerformanceProducer.class);
+
+ public static AtomicReference<Transaction> buildTransaction(PulsarClient pulsarClient, boolean isEnableTransaction,
+ long transactionTimeout) {
+
+ AtomicLong numBuildTxnFailed = new AtomicLong();
+ if (isEnableTransaction) {
+ while(true) {
+ AtomicReference atomicReference = null;
+ try {
+ atomicReference = new AtomicReference(pulsarClient.newTransaction()
+ .withTransactionTimeout(transactionTimeout, TimeUnit.SECONDS).build().get());
+ } catch (Exception e) {
+ numBuildTxnFailed.incrementAndGet();
+ if (numBuildTxnFailed.get()%10 == 0) {
+ log.error("Failed to new a transaction with {} times", numBuildTxnFailed.get(), e);
+ }
+ }
+ if (atomicReference != null && atomicReference.get() != null) {
+ log.info("After {} failures, the transaction was created successfully for the first time",
+ numBuildTxnFailed.get());
+ return atomicReference;
+ }
+ }
+ }
+ return new AtomicReference<>(null);
+ }
+}
diff --git a/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceTransactionTest.java b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceTransactionTest.java
new file mode 100644
index 0000000..e04ea04
--- /dev/null
+++ b/pulsar-testclient/src/test/java/org/apache/pulsar/testclient/PerformanceTransactionTest.java
@@ -0,0 +1,240 @@
+/**
+ * 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.testclient;
+
+import com.google.common.collect.Sets;
+import java.net.URL;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+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.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.testng.Assert.fail;
+
+@Slf4j
+public class PerformanceTransactionTest extends MockedPulsarServiceBaseTest {
+ private final String testTenant = "pulsar";
+ private final String testNamespace = "perf";
+ private final String myNamespace = testTenant + "/" + testNamespace;
+ private final String testTopic = "persistent://" + myNamespace + "/test-";
+ private final AtomicInteger lastExitCode = new AtomicInteger(0);
+
+ @BeforeMethod
+ @Override
+ protected void setup() throws Exception {
+ ServiceConfiguration serviceConfiguration = getDefaultConf();
+ serviceConfiguration.setSystemTopicEnabled(true);
+ serviceConfiguration.setTransactionCoordinatorEnabled(true);
+ super.internalSetup(serviceConfiguration);
+ PerfClientUtils.setExitProcedure(code -> {
+ log.error("JVM exit code is {}", code);
+ if (code != 0) {
+ throw new RuntimeException("JVM should exit with code " + code);
+ }
+ });
+ // Setup namespaces
+ admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
+ admin.tenants().createTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(),
+ new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
+ admin.namespaces().createNamespace(myNamespace, Sets.newHashSet("test"));
+ admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString());
+ admin.topics().createPartitionedTopic(TopicName.TRANSACTION_COORDINATOR_ASSIGN.toString(), 1);
+ }
+
+ @AfterMethod(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ int exitCode = lastExitCode.get();
+ if (exitCode != 0) {
+ fail("Unexpected JVM exit code "+exitCode);
+ }
+ }
+
+ @Test
+ public void testTxnPerf() throws Exception {
+ String argString = "--topics-c %s --topics-p %s -threads 1 -ntxn 50 -u %s -ss %s -np 1 -au %s";
+ String testConsumeTopic = testTopic + UUID.randomUUID().toString();
+ String testProduceTopic = testTopic + UUID.randomUUID().toString();
+ String testSub = "testSub";
+ admin.topics().createPartitionedTopic(testConsumeTopic, 1);
+ String args = String.format(argString, testConsumeTopic, testProduceTopic,
+ pulsar.getBrokerServiceUrl(), testSub, new URL(pulsar.getWebServiceAddress()));
+
+ PulsarClient pulsarClient = PulsarClient.builder()
+ .enableTransaction(true)
+ .serviceUrl(pulsar.getBrokerServiceUrl())
+ .connectionsPerBroker(100)
+ .statsInterval(0, TimeUnit.SECONDS)
+ .build();
+ Producer<byte[]> produceToConsumeTopic = pulsarClient.newProducer(Schema.BYTES)
+ .producerName("perf-transaction-producer")
+ .sendTimeout(0, TimeUnit.SECONDS)
+ .topic(testConsumeTopic)
+ .create();
+ pulsarClient.newConsumer(Schema.BYTES)
+ .consumerName("perf-transaction-consumeVerify")
+ .topic(testConsumeTopic)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName(testSub + "pre")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ CountDownLatch countDownLatch = new CountDownLatch(50);
+ for (int i = 0; i < 50
+ ; i++) {
+ produceToConsumeTopic.newMessage().value(("testConsume " + i).getBytes()).sendAsync().thenRun(() -> {
+ countDownLatch.countDown();
+ });
+ }
+
+ countDownLatch.await();
+
+ Thread thread = new Thread(() -> {
+ try {
+ PerformanceTransaction.main(args.split(" "));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+ thread.start();
+ thread.join();
+ Consumer<byte[]> consumeFromConsumeTopic = pulsarClient.newConsumer(Schema.BYTES)
+ .consumerName("perf-transaction-consumeVerify")
+ .topic(testConsumeTopic)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName(testSub)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ Consumer<byte[]> consumeFromProduceTopic = pulsarClient.newConsumer(Schema.BYTES)
+ .consumerName("perf-transaction-produceVerify")
+ .topic(testProduceTopic)
+ .subscriptionName(testSub)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ for (int i = 0; i < 50; i++) {
+ Message message = consumeFromProduceTopic.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ consumeFromProduceTopic.acknowledge(message);
+ }
+ Message message = consumeFromConsumeTopic.receive(2, TimeUnit.SECONDS);
+ Assert.assertNull(message);
+ message = consumeFromProduceTopic.receive(2, TimeUnit.SECONDS);
+ Assert.assertNull(message);
+ }
+
+
+ @Test
+ public void testProduceTxnMessage() throws InterruptedException, PulsarClientException {
+ String argString = "%s -r 10 -u %s -m %d -txn";
+ String topic = testTopic + UUID.randomUUID();
+ int totalMessage = 100;
+ String args = String.format(argString, topic, pulsar.getBrokerServiceUrl(), totalMessage);
+ pulsarClient.newConsumer().subscriptionName("subName" + "pre").topic(topic)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .enableBatchIndexAcknowledgment(false)
+ .subscribe();
+ Thread thread = new Thread(() -> {
+ try {
+ log.info("");
+ PerformanceProducer.main(args.split(" "));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+ thread.start();
+ thread.join();
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName("subName").topic(topic)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .enableBatchIndexAcknowledgment(false)
+ .subscribe();
+ for (int i = 0; i < totalMessage; i++) {
+ Message message = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ consumer.acknowledge(message);
+ }
+ Message message = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNull(message);
+ }
+
+ @Test
+ public void testConsumeTxnMessage() throws InterruptedException, PulsarClientException, ExecutionException {
+ String argString = "%s -r 10 -u %s -txn -ss %s -st %s -sp %s -ntxn %d";
+ String subName = "sub";
+ String topic = testTopic + UUID.randomUUID();
+ String args = String.format(argString, topic, pulsar.getBrokerServiceUrl(), subName,
+ SubscriptionType.Exclusive, SubscriptionInitialPosition.Earliest, 10);
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).sendTimeout(0, TimeUnit.SECONDS)
+ .create();
+ pulsarClient.newConsumer(Schema.BYTES)
+ .consumerName("perf-transaction-consumeVerify")
+ .topic(topic)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionName(subName + "pre")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ for (int i = 0; i < 505; i++) {
+ producer.newMessage().send();
+ }
+ Thread thread = new Thread(() -> {
+ try {
+ log.info("");
+ PerformanceConsumer.main(args.split(" "));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+ thread.start();
+ thread.join();
+ Consumer<byte[]> consumer = pulsarClient.newConsumer().subscriptionName(subName).topic(topic)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .enableBatchIndexAcknowledgment(false)
+ .subscribe();
+ for (int i = 0; i < 5; i++) {
+ Message message = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNotNull(message);
+ }
+ Message message = consumer.receive(2, TimeUnit.SECONDS);
+ Assert.assertNull(message);
+ }
+
+}
[pulsar] 08/18: [function] pulsar admin exposes secrets for function (#12950)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 13b07fdb8ef3a2f2f02d29c619a0f712cf9df15a
Author: Neng Lu <nl...@streamnative.io>
AuthorDate: Wed Nov 24 18:31:45 2021 -0800
[function] pulsar admin exposes secrets for function (#12950)
### Motivation
Fixes #12834
### Modifications
add the `--secrets` argument into `pulsar-admin functions create/update/localrun` command
(cherry picked from commit 9c0cee959e0f215a9a33adca98948f3a5fdf2dfa)
---
.../main/java/org/apache/pulsar/admin/cli/CmdFunctions.java | 12 ++++++++++++
1 file changed, 12 insertions(+)
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index 0664ca8..2efef99 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -37,6 +37,7 @@ import java.io.File;
import java.lang.reflect.Field;
import java.lang.reflect.Type;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -319,6 +320,8 @@ public class CmdFunctions extends CmdBase {
protected Integer maxMessageRetries;
@Parameter(names = "--custom-runtime-options", description = "A string that encodes options to customize the runtime, see docs for configured runtime for details")
protected String customRuntimeOptions;
+ @Parameter(names = "--secrets", description = "The map of secretName to an object that encapsulates how the secret is fetched by the underlying secrets provider")
+ protected String secretsString;
@Parameter(names = "--dead-letter-topic", description = "The topic where messages that are not processed successfully are sent to")
protected String deadLetterTopic;
protected FunctionConfig functionConfig;
@@ -520,6 +523,15 @@ public class CmdFunctions extends CmdBase {
functionConfig.setCustomRuntimeOptions(customRuntimeOptions);
}
+ if (secretsString != null) {
+ Type type = new TypeToken<Map<String, Object>>() {}.getType();
+ Map<String, Object> secretsMap = new Gson().fromJson(secretsString, type);
+ if (secretsMap == null) {
+ secretsMap = Collections.emptyMap();
+ }
+ functionConfig.setSecrets(secretsMap);
+ }
+
// window configs
WindowConfig windowConfig = functionConfig.getWindowConfig();
if (null != windowLengthCount) {
[pulsar] 10/18: [c++] Define and expose PULSAR_VERSION macro (#12769)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a5c34667e1eb293a928efded16d7de73c0add6ea
Author: Masahiro Sakamoto <ma...@yahoo-corp.jp>
AuthorDate: Fri Nov 26 11:43:13 2021 +0900
[c++] Define and expose PULSAR_VERSION macro (#12769)
* Define and expose PULSAR_VERSION macro
* Change default value of _PULSAR_VERSION_INTERNAL_ to unknown
(cherry picked from commit 2dcf7e924e3d393a23fca3522f539e95c5ae4d62)
---
pulsar-client-cpp/.gitignore | 3 ++
pulsar-client-cpp/CMakeLists.txt | 4 +++
.../{lib/Version.h => include/pulsar/c/version.h} | 8 ++---
pulsar-client-cpp/lib/CMakeLists.txt | 2 +-
pulsar-client-cpp/lib/Commands.cc | 6 ++--
pulsar-client-cpp/lib/HTTPLookupService.cc | 2 +-
pulsar-client-cpp/lib/HTTPLookupService.h | 2 +-
.../lib/{Version.h => VersionInternal.h} | 10 +++----
.../{lib/Version.h => templates/Version.h.in} | 14 +++++----
.../{lib/Version.h => tests/VersionTest.cc} | 15 ++++++----
src/gen-pulsar-version-macro.py | 35 ++++++++++++++++++++++
11 files changed, 72 insertions(+), 29 deletions(-)
diff --git a/pulsar-client-cpp/.gitignore b/pulsar-client-cpp/.gitignore
index 0f3d36d..f2a623d 100644
--- a/pulsar-client-cpp/.gitignore
+++ b/pulsar-client-cpp/.gitignore
@@ -47,6 +47,9 @@ lib*.so*
/perf/perfConsumer
/system-test/SystemTest
+# Files generated from templates by CMAKE
+include/pulsar/Version.h
+
# IDE generated files
.csettings
.cproject
diff --git a/pulsar-client-cpp/CMakeLists.txt b/pulsar-client-cpp/CMakeLists.txt
index 7c95791..8521dbf 100644
--- a/pulsar-client-cpp/CMakeLists.txt
+++ b/pulsar-client-cpp/CMakeLists.txt
@@ -21,6 +21,10 @@ cmake_minimum_required(VERSION 3.4)
project (pulsar-cpp)
set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake_modules")
+execute_process(COMMAND python ${CMAKE_SOURCE_DIR}/../src/gen-pulsar-version-macro.py OUTPUT_STRIP_TRAILING_WHITESPACE OUTPUT_VARIABLE PVM)
+set(PVM_COMMENT "This is generated from Version.h.in by CMAKE. DO NOT EDIT DIRECTLY")
+configure_file(templates/Version.h.in include/pulsar/Version.h @ONLY)
+
if (VCPKG_TRIPLET)
message(STATUS "Use vcpkg, triplet is ${VCPKG_TRIPLET}")
set(CMAKE_PREFIX_PATH "${CMAKE_SOURCE_DIR}/vcpkg_installed/${VCPKG_TRIPLET}")
diff --git a/pulsar-client-cpp/lib/Version.h b/pulsar-client-cpp/include/pulsar/c/version.h
similarity index 85%
copy from pulsar-client-cpp/lib/Version.h
copy to pulsar-client-cpp/include/pulsar/c/version.h
index a274f47..ab63c8a 100644
--- a/pulsar-client-cpp/lib/Version.h
+++ b/pulsar-client-cpp/include/pulsar/c/version.h
@@ -16,11 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-#ifndef LIB_VERSION_H_
-#define LIB_VERSION_H_
-#ifndef _PULSAR_VERSION_
-#define _PULSAR_VERSION_ "1.17"
-#endif
+#pragma once
-#endif /* LIB_VERSION_H_ */
+#include <pulsar/Version.h>
diff --git a/pulsar-client-cpp/lib/CMakeLists.txt b/pulsar-client-cpp/lib/CMakeLists.txt
index 6e970fa..ee9214e 100644
--- a/pulsar-client-cpp/lib/CMakeLists.txt
+++ b/pulsar-client-cpp/lib/CMakeLists.txt
@@ -20,7 +20,7 @@
file(GLOB PULSAR_SOURCES *.cc *.h lz4/*.cc lz4/*.h checksum/*.cc checksum/*.h stats/*.cc stats/*.h c/*.cc c/*.h auth/*.cc auth/*.h auth/athenz/*.cc auth/athenz/*.h)
execute_process(COMMAND python ${CMAKE_SOURCE_DIR}/../src/get-project-version.py OUTPUT_STRIP_TRAILING_WHITESPACE OUTPUT_VARIABLE PV)
-set (CMAKE_CXX_FLAGS " ${CMAKE_CXX_FLAGS} -D_PULSAR_VERSION_=\\\"${PV}\\\"")
+set (CMAKE_CXX_FLAGS " ${CMAKE_CXX_FLAGS} -D_PULSAR_VERSION_INTERNAL_=\\\"${PV}\\\"")
if (NOT PROTOC_PATH)
set(PROTOC_PATH protoc)
diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc
index 54c8c65..1094efb 100644
--- a/pulsar-client-cpp/lib/Commands.cc
+++ b/pulsar-client-cpp/lib/Commands.cc
@@ -18,7 +18,7 @@
*/
#include "Commands.h"
#include "MessageImpl.h"
-#include "Version.h"
+#include "VersionInternal.h"
#include "pulsar/MessageBuilder.h"
#include "LogUtils.h"
#include "PulsarApi.pb.h"
@@ -215,7 +215,7 @@ SharedBuffer Commands::newConnect(const AuthenticationPtr& authentication, const
BaseCommand cmd;
cmd.set_type(BaseCommand::CONNECT);
CommandConnect* connect = cmd.mutable_connect();
- connect->set_client_version(_PULSAR_VERSION_);
+ connect->set_client_version(_PULSAR_VERSION_INTERNAL_);
connect->set_auth_method_name(authentication->getAuthMethodName());
connect->set_protocol_version(ProtocolVersion_MAX);
@@ -243,7 +243,7 @@ SharedBuffer Commands::newAuthResponse(const AuthenticationPtr& authentication,
BaseCommand cmd;
cmd.set_type(BaseCommand::AUTH_RESPONSE);
CommandAuthResponse* authResponse = cmd.mutable_authresponse();
- authResponse->set_client_version(_PULSAR_VERSION_);
+ authResponse->set_client_version(_PULSAR_VERSION_INTERNAL_);
AuthData* authData = authResponse->mutable_response();
authData->set_auth_method_name(authentication->getAuthMethodName());
diff --git a/pulsar-client-cpp/lib/HTTPLookupService.cc b/pulsar-client-cpp/lib/HTTPLookupService.cc
index a54a4c1..d377171 100644
--- a/pulsar-client-cpp/lib/HTTPLookupService.cc
+++ b/pulsar-client-cpp/lib/HTTPLookupService.cc
@@ -151,7 +151,7 @@ void HTTPLookupService::handleNamespaceTopicsHTTPRequest(NamespaceTopicsPromise
Result HTTPLookupService::sendHTTPRequest(const std::string completeUrl, std::string &responseData) {
CURL *handle;
CURLcode res;
- std::string version = std::string("Pulsar-CPP-v") + _PULSAR_VERSION_;
+ std::string version = std::string("Pulsar-CPP-v") + _PULSAR_VERSION_INTERNAL_;
handle = curl_easy_init();
if (!handle) {
diff --git a/pulsar-client-cpp/lib/HTTPLookupService.h b/pulsar-client-cpp/lib/HTTPLookupService.h
index 166a14a..eb73654 100644
--- a/pulsar-client-cpp/lib/HTTPLookupService.h
+++ b/pulsar-client-cpp/lib/HTTPLookupService.h
@@ -22,7 +22,7 @@
#include <lib/LookupService.h>
#include <lib/ClientImpl.h>
#include <lib/Url.h>
-#include <lib/Version.h>
+#include <lib/VersionInternal.h>
namespace pulsar {
class HTTPLookupService : public LookupService, public std::enable_shared_from_this<HTTPLookupService> {
diff --git a/pulsar-client-cpp/lib/Version.h b/pulsar-client-cpp/lib/VersionInternal.h
similarity index 82%
copy from pulsar-client-cpp/lib/Version.h
copy to pulsar-client-cpp/lib/VersionInternal.h
index a274f47..c256035 100644
--- a/pulsar-client-cpp/lib/Version.h
+++ b/pulsar-client-cpp/lib/VersionInternal.h
@@ -16,11 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
-#ifndef LIB_VERSION_H_
-#define LIB_VERSION_H_
+#ifndef LIB_VERSION_INTERNAL_H_
+#define LIB_VERSION_INTERNAL_H_
-#ifndef _PULSAR_VERSION_
-#define _PULSAR_VERSION_ "1.17"
+#ifndef _PULSAR_VERSION_INTERNAL_
+#define _PULSAR_VERSION_INTERNAL_ "unknown"
#endif
-#endif /* LIB_VERSION_H_ */
+#endif /* LIB_VERSION_INTERNAL_H_ */
diff --git a/pulsar-client-cpp/lib/Version.h b/pulsar-client-cpp/templates/Version.h.in
similarity index 85%
copy from pulsar-client-cpp/lib/Version.h
copy to pulsar-client-cpp/templates/Version.h.in
index a274f47..d52121a 100644
--- a/pulsar-client-cpp/lib/Version.h
+++ b/pulsar-client-cpp/templates/Version.h.in
@@ -16,11 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-#ifndef LIB_VERSION_H_
-#define LIB_VERSION_H_
-#ifndef _PULSAR_VERSION_
-#define _PULSAR_VERSION_ "1.17"
-#endif
+/**
+ * @PVM_COMMENT@
+ */
+#ifndef PULSAR_VERSION_H_
+#define PULSAR_VERSION_H_
+
+#define PULSAR_VERSION @PVM@
-#endif /* LIB_VERSION_H_ */
+#endif /* PULSAR_VERSION_H_ */
diff --git a/pulsar-client-cpp/lib/Version.h b/pulsar-client-cpp/tests/VersionTest.cc
similarity index 79%
rename from pulsar-client-cpp/lib/Version.h
rename to pulsar-client-cpp/tests/VersionTest.cc
index a274f47..57e1e78 100644
--- a/pulsar-client-cpp/lib/Version.h
+++ b/pulsar-client-cpp/tests/VersionTest.cc
@@ -16,11 +16,14 @@
* specific language governing permissions and limitations
* under the License.
*/
-#ifndef LIB_VERSION_H_
-#define LIB_VERSION_H_
+#include <pulsar/Version.h>
+#include <gtest/gtest.h>
-#ifndef _PULSAR_VERSION_
-#define _PULSAR_VERSION_ "1.17"
+TEST(VersionTest, testMacro) {
+#ifdef PULSAR_VERSION
+ ASSERT_GE(PULSAR_VERSION, 2000000);
+ ASSERT_LE(PULSAR_VERSION, 999999999);
+#else
+ FAIL();
#endif
-
-#endif /* LIB_VERSION_H_ */
+}
diff --git a/src/gen-pulsar-version-macro.py b/src/gen-pulsar-version-macro.py
new file mode 100755
index 0000000..f32df91
--- /dev/null
+++ b/src/gen-pulsar-version-macro.py
@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+#
+# 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.
+#
+
+import xml.etree.ElementTree as ET
+import re
+from os.path import dirname, realpath, join
+
+# Derive the POM path from the current script location
+TOP_LEVEL_PATH = dirname(dirname(realpath(__file__)))
+POM_PATH = join(TOP_LEVEL_PATH, 'pom.xml')
+
+root = ET.XML(open(POM_PATH).read())
+m = re.search(r'^(\d+)\.(\d+)\.(\d+)', root.find('{http://maven.apache.org/POM/4.0.0}version').text)
+
+version_macro = 0
+for i in range(3):
+ version_macro += int(m.group(3 - i)) * (1000 ** i)
+print(version_macro)
[pulsar] 13/18: Fix update ledger list to znode version mismatch failed, ledger not delete (#12015)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 71ba8ef2afda1a71663d5b2e7f415ba73dc3ea2c
Author: Hang Chen <ch...@apache.org>
AuthorDate: Mon Nov 29 12:22:55 2021 +0800
Fix update ledger list to znode version mismatch failed, ledger not delete (#12015)
### Motivation
When Zookeeper throws `Failed to update ledger list. z-node version mismatch. Closing managed ledger` exception when update ZNode list, it will not clean up the created ledger, which will lead to the new created ledger not be indexed to the topic managedLedger list, and can't be cleanup as topic retention. What's more, it will cause ZNode number increase in Zookeeper if the `z-node version mismatch` exception keeping throw out. The exception list as follow:
```
10:44:29.017 [main-EventThread] INFO org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [test/test/persistent/test_v1-partition-4] Created new ledger 67311140
10:44:29.018 [bookkeeper-ml-workers-OrderedExecutor-2-0] ERROR org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [test/test/persistent/test_v1-partition-4] Failed to update ledger list. z-node version mismatch. Closing managed ledger
10:44:29.018 [bookkeeper-ml-workers-OrderedExecutor-2-0] INFO org.apache.pulsar.broker.service.Producer - Disconnecting producer: Producer{topic=PersistentTopic{topic=persistent://test/test/test_v1-partition-4}, client=/10.1.2.3:38938, producerName=pulsar-101-1123, producerId=20}
```
### Modification
1. When updating ZNode list failed, delete the created ledger from broker cache and BookKeeper, regardless of whether the exception type is BadVersionException or not.
(cherry picked from commit e7b0e3dbc5b6a05e955dc2cad034de0487463150)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 30 +++++++++++-----------
1 file changed, 15 insertions(+), 15 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index caff296..7f4d291 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -1438,22 +1438,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public void operationFailed(MetaStoreException e) {
- if (e instanceof BadVersionException) {
- synchronized (ManagedLedgerImpl.this) {
- log.error(
- "[{}] Failed to update ledger list. z-node version mismatch. Closing managed ledger",
- name);
- STATE_UPDATER.set(ManagedLedgerImpl.this, State.Fenced);
- // Return ManagedLedgerFencedException to addFailed callback
- // to indicate that the ledger is now fenced and topic needs to be closed
- clearPendingAddEntries(new ManagedLedgerFencedException(e));
- // Do not need to unlock ledgersListMutex here because we are going to close to topic anyways
- return;
- }
- }
-
log.warn("[{}] Error updating meta data with the new list of ledgers: {}", name, e.getMessage());
-
// Remove the ledger, since we failed to update the list
ledgers.remove(lh.getId());
mbean.startDataLedgerDeleteOp();
@@ -1465,6 +1450,21 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}, null);
+ if (e instanceof BadVersionException) {
+ synchronized (ManagedLedgerImpl.this) {
+ log.error(
+ "[{}] Failed to update ledger list. z-node version mismatch. Closing managed ledger",
+ name);
+ lastLedgerCreationFailureTimestamp = clock.millis();
+ STATE_UPDATER.set(ManagedLedgerImpl.this, State.Fenced);
+ // Return ManagedLedgerFencedException to addFailed callback
+ // to indicate that the ledger is now fenced and topic needs to be closed
+ clearPendingAddEntries(new ManagedLedgerFencedException(e));
+ // Do not need to unlock ledgersListMutex here because we are going to close to topic anyways
+ return;
+ }
+ }
+
metadataMutex.unlock();
synchronized (ManagedLedgerImpl.this) {
[pulsar] 16/18: Use sendAsync instead of send when produce message to retry topic. (#12946)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ab57a6a60854ca6dc225e53323adedc3c1322716
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Tue Nov 30 01:46:53 2021 +0800
Use sendAsync instead of send when produce message to retry topic. (#12946)
* Use sendAsync instead of send when produce message to retry letter topic.
* add exception handler.
(cherry picked from commit 09cc1d6aa91422c71601664dac8d94ba574beb7b)
---
.../main/java/org/apache/pulsar/client/impl/ConsumerImpl.java | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
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 1b8d19c..ff0b826 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
@@ -590,9 +590,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_RECONSUMETIMES, String.valueOf(reconsumetimes));
propertiesMap.put(RetryMessageUtil.SYSTEM_PROPERTY_DELAY_TIME, String.valueOf(unit.toMillis(delayTime)));
+ MessageId finalMessageId = messageId;
if (reconsumetimes > this.deadLetterPolicy.getMaxRedeliverCount() && StringUtils.isNotBlank(deadLetterPolicy.getDeadLetterTopic())) {
initDeadLetterProducerIfNeeded();
- MessageId finalMessageId = messageId;
deadLetterProducer.thenAccept(dlqProducer -> {
TypedMessageBuilder<byte[]> typedMessageBuilderNew =
dlqProducer.newMessage(Schema.AUTO_PRODUCE_BYTES(retryMessage.getReaderSchema().get()))
@@ -624,8 +624,12 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (message.hasKey()) {
typedMessageBuilderNew.key(message.getKey());
}
- typedMessageBuilderNew.send();
- return doAcknowledge(messageId, ackType, properties, null);
+ typedMessageBuilderNew.sendAsync()
+ .thenAccept(__ -> doAcknowledge(finalMessageId, ackType, properties, null).thenAccept(v -> result.complete(null)))
+ .exceptionally(ex -> {
+ result.completeExceptionally(ex);
+ return null;
+ });
}
} catch (Exception e) {
log.error("Send to retry letter topic exception with topic: {}, messageId: {}", retryLetterProducer.getTopic(), messageId, e);
[pulsar] 11/18: Fix NPE in checkSubscriptionTypesEnable (#12961)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 73863f8fb87d33f9d90191dfc2aae7d30fd3b676
Author: Zike Yang <ar...@armail.top>
AuthorDate: Fri Nov 26 11:23:12 2021 +0800
Fix NPE in checkSubscriptionTypesEnable (#12961)
The NPE may appear in `PersistentTopic.checkSubscriptionTypesEnable`. The `topicPolicies.getSubscriptionTypesEnabled()` may be null.
The problem occurred when upgrading from pulsar version 2.7 to 2.8. We added `SubscriptionTypesEnable` in 2.8: https://github.com/apache/pulsar/pull/9401. When we initialize a topic's topicPolicy in pulasr 2.7, and then get the topic's topicPolicy in version 2.8, the `SubscriptionTypesEnable` will be null. This leads to the problem.
Here are steps to reproduce:
* Start broker 2.7
* Create a topic and init the topic policy
* Upgrade broker to 2.8
* We will get the `SubscriptionTypesEnable` with the value of null
```sh
➜ pulsar-281 bin/pulsar-admin topics get-subscription-types-enabled my-topic
null
```
* When we consume from that topic, the issue occurs
```
Caused by: java.lang.NullPointerException
at org.apache.pulsar.broker.service.persistent.PersistentTopic.checkSubscriptionTypesEnable(PersistentTopic.java:3206) ~[io.streamnative-pulsar-broker-2.8.1.21.jar:2.8.1.21]
at org.apache.pulsar.broker.service.persistent.PersistentTopic.subscribe(PersistentTopic.java:688) ~[io.streamnative-pulsar-broker-2.8.1.21.jar:2.8.1.21]
at org.apache.pulsar.broker.service.ServerCnx.lambda$null$13(ServerCnx.java:1029) ~[io.streamnative-pulsar-broker-2.8.1.21.jar:2.8.1.21]
at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:1106) ~[?:?]
... 31 more
```
* Use `CollectionUtils.isEmpty` to determine if the list is empty to fix the problem of throwing NPE.
(cherry picked from commit 877bf3a34e72336c05706ffd48826e4347606196)
---
.../org/apache/pulsar/broker/service/persistent/PersistentTopic.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
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 f4adb2a..46c9dfd 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
@@ -72,6 +72,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.bookkeeper.net.BookieId;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -3198,7 +3199,7 @@ public class PersistentTopic extends AbstractTopic
if (topicPolicies == null) {
return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType);
} else {
- if (topicPolicies.getSubscriptionTypesEnabled().isEmpty()) {
+ if (CollectionUtils.isEmpty(topicPolicies.getSubscriptionTypesEnabled())) {
return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType);
}
return topicPolicies.getSubscriptionTypesEnabled().contains(subType);
[pulsar] 09/18: Fix wrong isEmpty method of ConcurrentOpenLongPairRangeSet (#12953)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8113c17ff68cad0f0ffa48fd2f31383699d3b355
Author: Yunze Xu <xy...@163.com>
AuthorDate: Thu Nov 25 10:29:41 2021 +0800
Fix wrong isEmpty method of ConcurrentOpenLongPairRangeSet (#12953)
(cherry picked from commit 6cc5cff0f051cf50c84edb4bb67034cb978d3646)
---
.../util/collections/ConcurrentOpenLongPairRangeSet.java | 12 +++++-------
.../collections/ConcurrentOpenLongPairRangeSetTest.java | 13 +++++++++++++
2 files changed, 18 insertions(+), 7 deletions(-)
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java
index 174f318..7ae6c7e 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSet.java
@@ -152,14 +152,12 @@ public class ConcurrentOpenLongPairRangeSet<T extends Comparable<T>> implements
if (rangeBitSetMap.isEmpty()) {
return true;
}
- AtomicBoolean isEmpty = new AtomicBoolean(false);
- rangeBitSetMap.forEach((key, val) -> {
- if (!isEmpty.get()) {
- return;
+ for (BitSet rangeBitSet : rangeBitSetMap.values()) {
+ if (!rangeBitSet.isEmpty()) {
+ return false;
}
- isEmpty.set(val.isEmpty());
- });
- return isEmpty.get();
+ }
+ return true;
}
@Override
diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java
index f57b75d..90c23ae 100644
--- a/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java
+++ b/pulsar-common/src/test/java/org/apache/pulsar/common/util/collections/ConcurrentOpenLongPairRangeSetTest.java
@@ -19,7 +19,9 @@
package org.apache.pulsar.common.util.collections;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
import java.util.List;
import java.util.Set;
@@ -38,6 +40,17 @@ public class ConcurrentOpenLongPairRangeSetTest {
static final LongPairConsumer<LongPair> consumer = (key, value) -> new LongPair(key, value);
@Test
+ public void testIsEmpty() {
+ ConcurrentOpenLongPairRangeSet<LongPair> set = new ConcurrentOpenLongPairRangeSet<>(consumer);
+ assertTrue(set.isEmpty());
+ // lowerValueOpen and upperValue are both -1 so that an empty set will be added
+ set.addOpenClosed(0, -1, 0, -1);
+ assertTrue(set.isEmpty());
+ set.addOpenClosed(1, 1, 1, 5);
+ assertFalse(set.isEmpty());
+ }
+
+ @Test
public void testAddForSameKey() {
ConcurrentOpenLongPairRangeSet<LongPair> set = new ConcurrentOpenLongPairRangeSet<>(consumer);
// add 0 to 5
[pulsar] 03/18: [Java Client] Avoid IllegalStateException in ClientCnx debug logs (#12899)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4954c09e87be2672ffe04f3a18487c7507484ed7
Author: Michael Marshall <mi...@datastax.com>
AuthorDate: Mon Nov 22 04:45:52 2021 -0500
[Java Client] Avoid IllegalStateException in ClientCnx debug logs (#12899)
(cherry picked from commit 32b697d3adb18ecc9992f6dfbf9ac13158649af3)
---
.../src/main/java/org/apache/pulsar/client/impl/ClientCnx.java | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index e2386a9..cae1594 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -543,7 +543,9 @@ public class ClientCnx extends PulsarHandler {
@Override
protected void handleLookupResponse(CommandLookupTopicResponse lookupResult) {
if (log.isDebugEnabled()) {
- log.debug("Received Broker lookup response: {}", lookupResult.getResponse());
+ CommandLookupTopicResponse.LookupType response =
+ lookupResult.hasResponse() ? lookupResult.getResponse() : null;
+ log.debug("Received Broker lookup response: {} {}", lookupResult.getRequestId(), response);
}
long requestId = lookupResult.getRequestId();
@@ -579,7 +581,11 @@ public class ClientCnx extends PulsarHandler {
@Override
protected void handlePartitionResponse(CommandPartitionedTopicMetadataResponse lookupResult) {
if (log.isDebugEnabled()) {
- log.debug("Received Broker Partition response: {}", lookupResult.getPartitions());
+ CommandPartitionedTopicMetadataResponse.LookupType response =
+ lookupResult.hasResponse() ? lookupResult.getResponse() : null;
+ int partitions = lookupResult.hasPartitions() ? lookupResult.getPartitions() : -1;
+ log.debug("Received Broker Partition response: {} {} {}", lookupResult.getRequestId(), response,
+ partitions);
}
long requestId = lookupResult.getRequestId();
[pulsar] 14/18: optimize if statement (#12741)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6b9d02cb0a488dfe84359b02785973e2e8a2f105
Author: gaozhangmin <ga...@qq.com>
AuthorDate: Tue Nov 30 03:51:54 2021 +0800
optimize if statement (#12741)
Co-authored-by: gavingaozhangmin <ga...@didiglobal.com>
(cherry picked from commit f3641a3fa89686705ebe4bf5ea9936af3003d465)
---
.../src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java | 6 +++---
.../src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java | 6 +++---
.../src/main/java/org/apache/pulsar/admin/cli/CmdSources.java | 6 +++---
.../src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java | 7 ++-----
.../java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java | 6 +-----
.../java/org/apache/pulsar/common/tls/TlsHostnameVerifier.java | 4 +---
.../java/org/apache/pulsar/functions/worker/FunctionActioner.java | 7 ++-----
.../src/main/java/org/apache/pulsar/io/flume/FlumeConnector.java | 5 +----
.../main/java/org/apache/pulsar/testclient/PerformanceReader.java | 2 +-
9 files changed, 17 insertions(+), 32 deletions(-)
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index 2efef99..90ebf68 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -711,13 +711,13 @@ public class CmdFunctions extends CmdBase {
if (isBlank(clientAuthParams) && !isBlank(DEPRECATED_clientAuthParams)) {
clientAuthParams = DEPRECATED_clientAuthParams;
}
- if (useTls == false && DEPRECATED_useTls != null) {
+ if (!useTls && DEPRECATED_useTls != null) {
useTls = DEPRECATED_useTls;
}
- if (tlsAllowInsecureConnection == false && DEPRECATED_tlsAllowInsecureConnection != null) {
+ if (!tlsAllowInsecureConnection && DEPRECATED_tlsAllowInsecureConnection != null) {
tlsAllowInsecureConnection = DEPRECATED_tlsAllowInsecureConnection;
}
- if (tlsHostNameVerificationEnabled == false && DEPRECATED_tlsHostNameVerificationEnabled != null) {
+ if (!tlsHostNameVerificationEnabled && DEPRECATED_tlsHostNameVerificationEnabled != null) {
tlsHostNameVerificationEnabled = DEPRECATED_tlsHostNameVerificationEnabled;
}
if (isBlank(tlsTrustCertFilePath) && !isBlank(DEPRECATED_tlsTrustCertFilePath)) {
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index a4bf82e..0035ff4 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -191,13 +191,13 @@ public class CmdSinks extends CmdBase {
if (isBlank(clientAuthParams) && !isBlank(DEPRECATED_clientAuthParams)) {
clientAuthParams = DEPRECATED_clientAuthParams;
}
- if (useTls == false && DEPRECATED_useTls != null) {
+ if (!useTls && DEPRECATED_useTls != null) {
useTls = DEPRECATED_useTls;
}
- if (tlsAllowInsecureConnection == false && DEPRECATED_tlsAllowInsecureConnection != null) {
+ if (!tlsAllowInsecureConnection && DEPRECATED_tlsAllowInsecureConnection != null) {
tlsAllowInsecureConnection = DEPRECATED_tlsAllowInsecureConnection;
}
- if (tlsHostNameVerificationEnabled == false && DEPRECATED_tlsHostNameVerificationEnabled != null) {
+ if (!tlsHostNameVerificationEnabled && DEPRECATED_tlsHostNameVerificationEnabled != null) {
tlsHostNameVerificationEnabled = DEPRECATED_tlsHostNameVerificationEnabled;
}
if (isBlank(tlsTrustCertFilePath) && !isBlank(DEPRECATED_tlsTrustCertFilePath)) {
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
index 6d836e2..f78feb0 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java
@@ -190,13 +190,13 @@ public class CmdSources extends CmdBase {
if (isBlank(clientAuthParams) && !isBlank(DEPRECATED_clientAuthParams)) {
clientAuthParams = DEPRECATED_clientAuthParams;
}
- if (useTls == false && DEPRECATED_useTls != null) {
+ if (!useTls && DEPRECATED_useTls != null) {
useTls = DEPRECATED_useTls;
}
- if (tlsAllowInsecureConnection == false && DEPRECATED_tlsAllowInsecureConnection != null) {
+ if (!tlsAllowInsecureConnection && DEPRECATED_tlsAllowInsecureConnection != null) {
tlsAllowInsecureConnection = DEPRECATED_tlsAllowInsecureConnection;
}
- if (tlsHostNameVerificationEnabled == false && DEPRECATED_tlsHostNameVerificationEnabled != null) {
+ if (!tlsHostNameVerificationEnabled && DEPRECATED_tlsHostNameVerificationEnabled != null) {
tlsHostNameVerificationEnabled = DEPRECATED_tlsHostNameVerificationEnabled;
}
if (isBlank(tlsTrustCertFilePath) && !isBlank(DEPRECATED_tlsTrustCertFilePath)) {
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 9312df5..1251593 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
@@ -211,13 +211,10 @@ public abstract class ConsumerBase<T> extends HandlerState implements Consumer<T
protected CompletableFuture<Message<T>> nextPendingReceive() {
CompletableFuture<Message<T>> receivedFuture;
- while (true) {
+ do {
receivedFuture = pendingReceives.poll();
// skip done futures (cancelling a future could mark it done)
- if (receivedFuture == null || !receivedFuture.isDone()) {
- break;
- }
- }
+ } while (receivedFuture != null && receivedFuture.isDone());
return receivedFuture;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java
index ce68298..7665d96 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ByteBufSchema.java
@@ -64,11 +64,7 @@ public class ByteBufSchema extends AbstractSchema<ByteBuf> {
@Override
public ByteBuf decode(ByteBuf byteBuf) {
- if (null == byteBuf) {
- return null;
- } else {
- return byteBuf;
- }
+ return byteBuf;
}
@Override
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/tls/TlsHostnameVerifier.java b/pulsar-common/src/main/java/org/apache/pulsar/common/tls/TlsHostnameVerifier.java
index 3ab5b5f..0735013 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/tls/TlsHostnameVerifier.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/tls/TlsHostnameVerifier.java
@@ -206,9 +206,7 @@ public class TlsHostnameVerifier implements HostnameVerifier {
if (strict) {
final String remainder = host.substring(
prefix.length(), host.length() - suffix.length());
- if (remainder.contains(".")) {
- return false;
- }
+ return !remainder.contains(".");
}
return true;
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
index 6497e15..5fa5540 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -204,14 +204,11 @@ public class FunctionActioner {
}
File tempPkgFile;
- while (true) {
+ do {
tempPkgFile = new File(
pkgDir,
pkgFile.getName() + "." + instanceId + "." + UUID.randomUUID().toString());
- if (!tempPkgFile.exists() && tempPkgFile.createNewFile()) {
- break;
- }
- }
+ } while (tempPkgFile.exists() || !tempPkgFile.createNewFile());
String pkgLocationPath = functionMetaData.getPackageLocation().getPackagePath();
boolean downloadFromHttp = isPkgUrlProvided && pkgLocationPath.startsWith(HTTP);
log.info("{}/{}/{} Function package file {} will be downloaded from {}", tempPkgFile, details.getTenant(),
diff --git a/pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/FlumeConnector.java b/pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/FlumeConnector.java
index 29a9847..66eb2d6 100644
--- a/pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/FlumeConnector.java
+++ b/pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/FlumeConnector.java
@@ -43,10 +43,7 @@ public class FlumeConnector {
SSLUtil.initGlobalSSLParameters();
String agentName = flumeConfig.getName();
boolean reload = !flumeConfig.getNoReloadConf();
- boolean isZkConfigured = false;
- if (flumeConfig.getZkConnString().length() > 0) {
- isZkConfigured = true;
- }
+ boolean isZkConfigured = flumeConfig.getZkConnString().length() > 0;
if (isZkConfigured) {
// get options
String zkConnectionStr = flumeConfig.getZkConnString();
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
index 44e5555..d18c76a 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
@@ -199,7 +199,7 @@ public class PerformanceReader {
arguments.authParams = prop.getProperty("authParams", null);
}
- if (arguments.useTls == false) {
+ if (!arguments.useTls) {
arguments.useTls = Boolean.parseBoolean(prop.getProperty("useTls"));
}
[pulsar] 04/18: [broker] Fix topic policy listener deleted by mistake. (#12904)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 96238e0eb1fcde8a8b064d5e710e76b0bfa5d758
Author: JiangHaiting <ji...@apache.org>
AuthorDate: Mon Nov 22 09:57:46 2021 +0800
[broker] Fix topic policy listener deleted by mistake. (#12904)
### Motivation
Here is the current way of dealing topic policy listeners in PersistentTopic, for example topic name is "A", with 3 partitions.
- Register: call TopicPoliciesService.registerListener("A", listener), for all 3 partitions of topic "A".
- Clean: call TopicPoliciesService.clean("A-partition-x"), here is the problem it will delete all listeners of all partitions of topic "A", if any partition is closed.
This means, if we calls `admin.topics().unload("A-partition-0")`, "A-partition-1" and "A-partition-2" will not be able to receive topic policy update callbacks any more.
A detailed case is designed in the new unit test `testListenerCleanupByPartition`.
### Modifications
With previous optimization of #12654 , now we can use `org.apache.pulsar.broker.service.TopicPoliciesService#unregisterListener` to do the clean up.
(cherry picked from commit a0c96a08de83de6ce51fffd06e907833f075bbca)
---
.../broker/service/persistent/PersistentTopic.java | 20 ++++++++++--------
.../SystemTopicBasedTopicPoliciesServiceTest.java | 24 ++++++++++++++++++++++
.../org/apache/pulsar/common/naming/TopicName.java | 8 ++++++++
3 files changed, 43 insertions(+), 9 deletions(-)
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 275c7ea..f4adb2a 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
@@ -1156,8 +1156,7 @@ public class PersistentTopic extends AbstractTopic
subscribeRateLimiter.ifPresent(SubscribeRateLimiter::close);
- brokerService.pulsar().getTopicPoliciesService()
- .clean(TopicName.get(topic));
+ unregisterTopicPolicyListener();
log.info("[{}] Topic deleted", topic);
deleteFuture.complete(null);
@@ -1263,7 +1262,7 @@ public class PersistentTopic extends AbstractTopic
subscribeRateLimiter.ifPresent(SubscribeRateLimiter::close);
- brokerService.pulsar().getTopicPoliciesService().clean(TopicName.get(topic));
+ unregisterTopicPolicyListener();
log.info("[{}] Topic closed", topic);
closeFuture.complete(null);
})
@@ -3148,13 +3147,16 @@ public class PersistentTopic extends AbstractTopic
private void registerTopicPolicyListener() {
if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
&& brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
- TopicName topicName = TopicName.get(topic);
- TopicName cloneTopicName = topicName;
- if (topicName.isPartitioned()) {
- cloneTopicName = TopicName.get(topicName.getPartitionedTopicName());
- }
+ brokerService.getPulsar().getTopicPoliciesService()
+ .registerListener(TopicName.getPartitionedTopicName(topic), this);
+ }
+ }
- brokerService.getPulsar().getTopicPoliciesService().registerListener(cloneTopicName, this);
+ private void unregisterTopicPolicyListener() {
+ if (brokerService.pulsar().getConfig().isSystemTopicEnabled()
+ && brokerService.pulsar().getConfig().isTopicLevelPoliciesEnabled()) {
+ brokerService.getPulsar().getTopicPoliciesService()
+ .unregisterListener(TopicName.getPartitionedTopicName(topic), this);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
index be52f9a..9a489cf 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
@@ -259,6 +259,30 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic
assertNull(listMap.get(topicName));
}
+ @Test
+ public void testListenerCleanupByPartition() throws Exception {
+ final String topic = "persistent://" + NAMESPACE1 + "/test" + UUID.randomUUID();
+ TopicName topicName = TopicName.get(topic);
+ admin.topics().createPartitionedTopic(topic, 3);
+ pulsarClient.newProducer().topic(topic).create().close();
+
+ Map<TopicName, List<TopicPolicyListener<TopicPolicies>>> listMap =
+ systemTopicBasedTopicPoliciesService.getListeners();
+ Awaitility.await().untilAsserted(() -> {
+ // all 3 topic partition have registered the topic policy listeners.
+ assertEquals(listMap.get(topicName).size(), 3);
+ });
+
+ admin.topics().unload(topicName.getPartition(0).toString());
+ assertEquals(listMap.get(topicName).size(), 2);
+ admin.topics().unload(topicName.getPartition(1).toString());
+ assertEquals(listMap.get(topicName).size(), 1);
+ admin.topics().unload(topicName.getPartition(2).toString());
+ assertNull(listMap.get(topicName));
+ }
+
+
+
private void prepareData() throws PulsarAdminException {
admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
admin.tenants().createTenant("system-topic",
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
index 3729697..67bc92a 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -90,6 +90,14 @@ public class TopicName implements ServiceUnitId {
}
}
+ public static TopicName getPartitionedTopicName(String topic) {
+ TopicName topicName = TopicName.get(topic);
+ if (topicName.isPartitioned()) {
+ return TopicName.get(topicName.getPartitionedTopicName());
+ }
+ return topicName;
+ }
+
public static boolean isValid(String topic) {
try {
get(topic);
[pulsar] 18/18: add missed import (#13037)
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.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8e689260abf47fa48520da6d36bdc8252bf008dc
Author: Aloys <lo...@gmail.com>
AuthorDate: Tue Nov 30 14:38:17 2021 +0800
add missed import (#13037)
### Motivation
The current master branch was broken and can't be compiled successfully due to there missing an import for Sets.
https://github.com/apache/pulsar/blob/master/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java#L220
### Modifications
add missed import
(cherry picked from commit bebfc772456e689b14f6d10a6231165993e8ab14)
---
.../src/main/java/org/apache/pulsar/admin/cli/CliCommand.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java
index 2f1cf72..d91a880 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CliCommand.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import com.google.common.collect.Sets;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.impl.MessageIdImpl;