You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/05/24 01:58:17 UTC
[pulsar] branch branch-2.11 updated (fc388242c83 -> 5a66d6cc7bb)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a change to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from fc388242c83 [fix][test] Use delta when comparing doubles in checkLoadReportNicSpeed (#20343)
new 9d813d0e3f0 [improve][fn] Use functions classloader in TopicSchema.newSchemaInstance() to fix ClassNotFoundException when using custom SerDe classes. (targeted for master) (#20115)
new c71f4c431b3 [fix][broker]Fix deadlock of metadata store (#20189)
new 5a66d6cc7bb [fix][broker] Fix class name typo `PrecisPublishLimiter` to "Precise" (#20310)
The 3 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:
.../pulsar/broker/namespace/NamespaceService.java | 32 +++++++++++---------
.../pulsar/broker/namespace/OwnershipCache.java | 6 ++--
.../pulsar/broker/service/AbstractTopic.java | 2 +-
...lishLimiter.java => PrecisePublishLimiter.java} | 10 +++----
.../PrecisTopicPublishRateThrottleTest.java | 2 +-
...terTest.java => PrecisePublishLimiterTest.java} | 30 +++++++++----------
.../broker/service/PublishRateLimiterTest.java | 34 ++++++++++++----------
.../pulsar/functions/instance/ContextImpl.java | 2 +-
.../apache/pulsar/functions/sink/PulsarSink.java | 2 +-
.../pulsar/functions/source/PulsarSource.java | 6 ++--
.../source/SingleConsumerPulsarSource.java | 11 +++----
.../pulsar/functions/source/TopicSchema.java | 15 ++++++++--
.../pulsar/functions/source/TopicSchemaTest.java | 8 +++--
.../integration/functions/PulsarFunctionsTest.java | 8 ++---
.../functions/PulsarFunctionsTestBase.java | 2 +-
.../functions/java/PulsarFunctionsJavaTest.java | 11 ++++---
.../java/PulsarWorkerRebalanceDrainTest.java | 15 ++++++----
.../functions/utils/CommandGenerator.java | 10 +++----
18 files changed, 113 insertions(+), 93 deletions(-)
rename pulsar-broker/src/main/java/org/apache/pulsar/broker/service/{PrecisPublishLimiter.java => PrecisePublishLimiter.java} (93%)
rename pulsar-broker/src/test/java/org/apache/pulsar/broker/service/{PrecisPublishLimiterTest.java => PrecisePublishLimiterTest.java} (57%)
[pulsar] 02/03: [fix][broker]Fix deadlock of metadata store (#20189)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c71f4c431b3d07baed9d1f77acf573e66dcb48c4
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu May 18 11:15:53 2023 +0800
[fix][broker]Fix deadlock of metadata store (#20189)
Motivation: This task loadOrCreatePersistentTopic occupied the event thread of the ZK client so that other ZK tasks could not be finished anymore(Including the task itself), and it calls bundlesCache.synchronous().get(nsname) which is a blocking method.
Modification: Since the method getBundle(topic) will eventually call the method bundlesCache.synchronous().get(nsname), use getBundleAsync(topic) instead of getBundle(topic) to avoid blocking the thread.
---
.../pulsar/broker/namespace/NamespaceService.java | 32 ++++++++++++----------
.../pulsar/broker/namespace/OwnershipCache.java | 6 ++--
2 files changed, 21 insertions(+), 17 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
index 700819e3f2a..4b2df9c5487 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -38,7 +38,9 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
@@ -1009,26 +1011,28 @@ public class NamespaceService implements AutoCloseable {
new IllegalArgumentException("Invalid class of NamespaceBundle: " + suName.getClass().getName()));
}
+ /**
+ * @Deprecated This method is only used in test now.
+ */
+ @Deprecated
public boolean isServiceUnitActive(TopicName topicName) {
try {
- OwnedBundle ownedBundle = ownershipCache.getOwnedBundle(getBundle(topicName));
- if (ownedBundle == null) {
- return false;
- }
- return ownedBundle.isActive();
- } catch (Exception e) {
- LOG.warn("Unable to find OwnedBundle for topic - [{}]", topicName, e);
- return false;
+ return isServiceUnitActiveAsync(topicName).get(pulsar.getConfig()
+ .getMetadataStoreOperationTimeoutSeconds(), SECONDS);
+ } catch (InterruptedException | ExecutionException | TimeoutException e) {
+ LOG.warn("Unable to find OwnedBundle for topic in time - [{}]", topicName, e);
+ throw new RuntimeException(e);
}
}
public CompletableFuture<Boolean> isServiceUnitActiveAsync(TopicName topicName) {
- Optional<CompletableFuture<OwnedBundle>> res = ownershipCache.getOwnedBundleAsync(getBundle(topicName));
- if (!res.isPresent()) {
- return CompletableFuture.completedFuture(false);
- }
-
- return res.get().thenApply(ob -> ob != null && ob.isActive());
+ return getBundleAsync(topicName).thenCompose(bundle -> {
+ Optional<CompletableFuture<OwnedBundle>> optionalFuture = ownershipCache.getOwnedBundleAsync(bundle);
+ if (!optionalFuture.isPresent()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ return optionalFuture.get().thenApply(ob -> ob != null && ob.isActive());
+ });
}
private boolean isNamespaceOwned(NamespaceName fqnn) throws Exception {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
index a6ba5fb5d90..05e89e78803 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
@@ -295,10 +295,10 @@ public class OwnershipCache {
/**
* Disable bundle in local cache and on zk.
- *
- * @param bundle
- * @throws Exception
+ * @Deprecated This is a dangerous method which is currently only used for test, it will occupy the ZK thread.
+ * Please switch to your own thread after calling this method.
*/
+ @Deprecated
public CompletableFuture<Void> disableOwnership(NamespaceBundle bundle) {
return updateBundleState(bundle, false)
.thenCompose(__ -> {
[pulsar] 03/03: [fix][broker] Fix class name typo `PrecisPublishLimiter` to "Precise" (#20310)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5a66d6cc7bb9920a3025b47e71bf4faeb57529e3
Author: Kim, Joo Hyuk <be...@gmail.com>
AuthorDate: Mon May 15 16:29:50 2023 +0900
[fix][broker] Fix class name typo `PrecisPublishLimiter` to "Precise" (#20310)
---
.../pulsar/broker/service/AbstractTopic.java | 2 +-
...lishLimiter.java => PrecisePublishLimiter.java} | 10 +++----
.../PrecisTopicPublishRateThrottleTest.java | 2 +-
...terTest.java => PrecisePublishLimiterTest.java} | 30 +++++++++----------
.../broker/service/PublishRateLimiterTest.java | 34 ++++++++++++----------
5 files changed, 40 insertions(+), 38 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 8a76f615189..fa27f87f99d 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -1238,7 +1238,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
|| this.topicPublishRateLimiter == PublishRateLimiter.DISABLED_RATE_LIMITER) {
// create new rateLimiter if rate-limiter is disabled
if (preciseTopicPublishRateLimitingEnable) {
- this.topicPublishRateLimiter = new PrecisPublishLimiter(publishRate,
+ this.topicPublishRateLimiter = new PrecisePublishLimiter(publishRate,
() -> this.enableCnxAutoRead(), brokerService.pulsar().getExecutor());
} else {
this.topicPublishRateLimiter = new PublishRateLimiterImpl(publishRate);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisPublishLimiter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisePublishLimiter.java
similarity index 93%
rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisPublishLimiter.java
rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisePublishLimiter.java
index 67cc46d95fa..55e24ab158e 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisPublishLimiter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PrecisePublishLimiter.java
@@ -24,7 +24,7 @@ import org.apache.pulsar.common.policies.data.PublishRate;
import org.apache.pulsar.common.util.RateLimitFunction;
import org.apache.pulsar.common.util.RateLimiter;
-public class PrecisPublishLimiter implements PublishRateLimiter {
+public class PrecisePublishLimiter implements PublishRateLimiter {
protected volatile int publishMaxMessageRate = 0;
protected volatile long publishMaxByteRate = 0;
// precise mode for publish rate limiter
@@ -33,18 +33,18 @@ public class PrecisPublishLimiter implements PublishRateLimiter {
private final RateLimitFunction rateLimitFunction;
private final ScheduledExecutorService scheduledExecutorService;
- public PrecisPublishLimiter(Policies policies, String clusterName, RateLimitFunction rateLimitFunction) {
+ public PrecisePublishLimiter(Policies policies, String clusterName, RateLimitFunction rateLimitFunction) {
this.rateLimitFunction = rateLimitFunction;
update(policies, clusterName);
this.scheduledExecutorService = null;
}
- public PrecisPublishLimiter(PublishRate publishRate, RateLimitFunction rateLimitFunction) {
+ public PrecisePublishLimiter(PublishRate publishRate, RateLimitFunction rateLimitFunction) {
this(publishRate, rateLimitFunction, null);
}
- public PrecisPublishLimiter(PublishRate publishRate, RateLimitFunction rateLimitFunction,
- ScheduledExecutorService scheduledExecutorService) {
+ public PrecisePublishLimiter(PublishRate publishRate, RateLimitFunction rateLimitFunction,
+ ScheduledExecutorService scheduledExecutorService) {
this.rateLimitFunction = rateLimitFunction;
update(publishRate);
this.scheduledExecutorService = scheduledExecutorService;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisTopicPublishRateThrottleTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisTopicPublishRateThrottleTest.java
index 7a7cc091b3f..06c32a58b8c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisTopicPublishRateThrottleTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisTopicPublishRateThrottleTest.java
@@ -164,7 +164,7 @@ public class PrecisTopicPublishRateThrottleTest extends BrokerTestBase{
"" + rateInMsg));
Topic topicRef = pulsar.getBrokerService().getTopicReference(topic).get();
Assert.assertNotNull(topicRef);
- PrecisPublishLimiter limiter = ((PrecisPublishLimiter) ((AbstractTopic) topicRef).topicPublishRateLimiter);
+ PrecisePublishLimiter limiter = ((PrecisePublishLimiter) ((AbstractTopic) topicRef).topicPublishRateLimiter);
Awaitility.await().untilAsserted(() -> Assert.assertEquals(limiter.publishMaxMessageRate, rateInMsg));
Assert.assertEquals(limiter.publishMaxByteRate, 0);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisPublishLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisePublishLimiterTest.java
similarity index 57%
rename from pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisPublishLimiterTest.java
rename to pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisePublishLimiterTest.java
index 61804e7255d..2becd5d7e10 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisPublishLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PrecisePublishLimiterTest.java
@@ -23,35 +23,35 @@ import static org.testng.Assert.assertTrue;
import org.apache.pulsar.common.policies.data.PublishRate;
import org.testng.annotations.Test;
-public class PrecisPublishLimiterTest {
+public class PrecisePublishLimiterTest {
@Test
void shouldResetMsgLimitAfterUpdate() {
- PrecisPublishLimiter precisPublishLimiter = new PrecisPublishLimiter(new PublishRate(), () -> {
+ PrecisePublishLimiter precisePublishLimiter = new PrecisePublishLimiter(new PublishRate(), () -> {
});
- precisPublishLimiter.update(new PublishRate(1, 1));
- assertFalse(precisPublishLimiter.tryAcquire(99, 99));
- precisPublishLimiter.update(new PublishRate(-1, 100));
- assertTrue(precisPublishLimiter.tryAcquire(99, 99));
+ precisePublishLimiter.update(new PublishRate(1, 1));
+ assertFalse(precisePublishLimiter.tryAcquire(99, 99));
+ precisePublishLimiter.update(new PublishRate(-1, 100));
+ assertTrue(precisePublishLimiter.tryAcquire(99, 99));
}
@Test
void shouldResetBytesLimitAfterUpdate() {
- PrecisPublishLimiter precisPublishLimiter = new PrecisPublishLimiter(new PublishRate(), () -> {
+ PrecisePublishLimiter precisePublishLimiter = new PrecisePublishLimiter(new PublishRate(), () -> {
});
- precisPublishLimiter.update(new PublishRate(1, 1));
- assertFalse(precisPublishLimiter.tryAcquire(99, 99));
- precisPublishLimiter.update(new PublishRate(100, -1));
- assertTrue(precisPublishLimiter.tryAcquire(99, 99));
+ precisePublishLimiter.update(new PublishRate(1, 1));
+ assertFalse(precisePublishLimiter.tryAcquire(99, 99));
+ precisePublishLimiter.update(new PublishRate(100, -1));
+ assertTrue(precisePublishLimiter.tryAcquire(99, 99));
}
@Test
void shouldCloseResources() throws Exception {
for (int i = 0; i < 20000; i++) {
- PrecisPublishLimiter precisPublishLimiter = new PrecisPublishLimiter(new PublishRate(100, 100), () -> {
+ PrecisePublishLimiter precisePublishLimiter = new PrecisePublishLimiter(new PublishRate(100, 100), () -> {
});
- precisPublishLimiter.tryAcquire(99, 99);
- precisPublishLimiter.close();
+ precisePublishLimiter.tryAcquire(99, 99);
+ precisePublishLimiter.close();
}
}
-}
\ No newline at end of file
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java
index ae65ee71a18..ea68382b49f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PublishRateLimiterTest.java
@@ -39,7 +39,7 @@ public class PublishRateLimiterTest {
private final PublishRate publishRate = new PublishRate(10, 100);
private final PublishRate newPublishRate = new PublishRate(20, 200);
- private PrecisPublishLimiter precisPublishLimiter;
+ private PrecisePublishLimiter precisePublishLimiter;
private PublishRateLimiterImpl publishRateLimiter;
@BeforeMethod
@@ -47,7 +47,7 @@ public class PublishRateLimiterTest {
policies.publishMaxMessageRate = new HashMap<>();
policies.publishMaxMessageRate.put(CLUSTER_NAME, publishRate);
- precisPublishLimiter = new PrecisPublishLimiter(policies, CLUSTER_NAME, () -> System.out.print("Refresh permit"));
+ precisePublishLimiter = new PrecisePublishLimiter(policies, CLUSTER_NAME, () -> System.out.print("Refresh permit"));
publishRateLimiter = new PublishRateLimiterImpl(policies, CLUSTER_NAME);
}
@@ -88,23 +88,25 @@ public class PublishRateLimiterTest {
@Test
public void testPrecisePublishRateLimiterUpdate() {
- assertFalse(precisPublishLimiter.tryAcquire(15, 150));
+ assertFalse(precisePublishLimiter.tryAcquire(15, 150));
//update
- precisPublishLimiter.update(newPublishRate);
- assertTrue(precisPublishLimiter.tryAcquire(15, 150));
+ precisePublishLimiter.update(newPublishRate);
+ assertTrue(precisePublishLimiter.tryAcquire(15, 150));
}
@Test
public void testPrecisePublishRateLimiterAcquire() throws Exception {
- Class precisPublishLimiterClass = Class.forName("org.apache.pulsar.broker.service.PrecisPublishLimiter");
- Field topicPublishRateLimiterOnMessageField = precisPublishLimiterClass.getDeclaredField("topicPublishRateLimiterOnMessage");
- Field topicPublishRateLimiterOnByteField = precisPublishLimiterClass.getDeclaredField("topicPublishRateLimiterOnByte");
+ Class precisePublishLimiterClass = Class.forName("org.apache.pulsar.broker.service.PrecisePublishLimiter");
+ Field topicPublishRateLimiterOnMessageField = precisePublishLimiterClass.getDeclaredField("topicPublishRateLimiterOnMessage");
+ Field topicPublishRateLimiterOnByteField = precisePublishLimiterClass.getDeclaredField("topicPublishRateLimiterOnByte");
topicPublishRateLimiterOnMessageField.setAccessible(true);
topicPublishRateLimiterOnByteField.setAccessible(true);
- RateLimiter topicPublishRateLimiterOnMessage = (RateLimiter)topicPublishRateLimiterOnMessageField.get(precisPublishLimiter);
- RateLimiter topicPublishRateLimiterOnByte = (RateLimiter)topicPublishRateLimiterOnByteField.get(precisPublishLimiter);
+ RateLimiter topicPublishRateLimiterOnMessage = (RateLimiter)topicPublishRateLimiterOnMessageField.get(
+ precisePublishLimiter);
+ RateLimiter topicPublishRateLimiterOnByte = (RateLimiter)topicPublishRateLimiterOnByteField.get(
+ precisePublishLimiter);
Method renewTopicPublishRateLimiterOnMessageMethod = topicPublishRateLimiterOnMessage.getClass().getDeclaredMethod("renew", null);
Method renewTopicPublishRateLimiterOnByteMethod = topicPublishRateLimiterOnByte.getClass().getDeclaredMethod("renew", null);
@@ -112,7 +114,7 @@ public class PublishRateLimiterTest {
renewTopicPublishRateLimiterOnByteMethod.setAccessible(true);
// running tryAcquire in order to lazyInit the renewTask
- precisPublishLimiter.tryAcquire(1, 10);
+ precisePublishLimiter.tryAcquire(1, 10);
Field onMessageRenewTaskField = topicPublishRateLimiterOnMessage.getClass().getDeclaredField("renewTask");
Field onByteRenewTaskField = topicPublishRateLimiterOnByte.getClass().getDeclaredField("renewTask");
@@ -129,30 +131,30 @@ public class PublishRateLimiterTest {
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
// tryAcquire not exceeded
- assertTrue(precisPublishLimiter.tryAcquire(1, 10));
+ assertTrue(precisePublishLimiter.tryAcquire(1, 10));
renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
// tryAcquire numOfMessages exceeded
- assertFalse(precisPublishLimiter.tryAcquire(11, 100));
+ assertFalse(precisePublishLimiter.tryAcquire(11, 100));
renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
// tryAcquire msgSizeInBytes exceeded
- assertFalse(precisPublishLimiter.tryAcquire(10, 101));
+ assertFalse(precisePublishLimiter.tryAcquire(10, 101));
renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
// tryAcquire exceeded exactly
- assertFalse(precisPublishLimiter.tryAcquire(10, 100));
+ assertFalse(precisePublishLimiter.tryAcquire(10, 100));
renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
renewTopicPublishRateLimiterOnMessageMethod.invoke(topicPublishRateLimiterOnMessage);
renewTopicPublishRateLimiterOnByteMethod.invoke(topicPublishRateLimiterOnByte);
// tryAcquire not exceeded
- assertTrue(precisPublishLimiter.tryAcquire(9, 99));
+ assertTrue(precisePublishLimiter.tryAcquire(9, 99));
}
}
[pulsar] 01/03: [improve][fn] Use functions classloader in TopicSchema.newSchemaInstance() to fix ClassNotFoundException when using custom SerDe classes. (targeted for master) (#20115)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9d813d0e3f04f683afe95aef2937b24a52d6aacd
Author: Gabriel Miklós <ga...@getbridge.com>
AuthorDate: Thu May 18 17:23:55 2023 +0200
[improve][fn] Use functions classloader in TopicSchema.newSchemaInstance() to fix ClassNotFoundException when using custom SerDe classes. (targeted for master) (#20115)
---
.../org/apache/pulsar/functions/instance/ContextImpl.java | 2 +-
.../java/org/apache/pulsar/functions/sink/PulsarSink.java | 2 +-
.../org/apache/pulsar/functions/source/PulsarSource.java | 6 +++---
.../functions/source/SingleConsumerPulsarSource.java | 11 ++++-------
.../org/apache/pulsar/functions/source/TopicSchema.java | 15 ++++++++++++---
.../apache/pulsar/functions/source/TopicSchemaTest.java | 8 +++++---
.../tests/integration/functions/PulsarFunctionsTest.java | 8 ++++----
.../integration/functions/PulsarFunctionsTestBase.java | 2 +-
.../functions/java/PulsarFunctionsJavaTest.java | 11 +++++++----
.../functions/java/PulsarWorkerRebalanceDrainTest.java | 15 +++++++++------
.../integration/functions/utils/CommandGenerator.java | 10 +++++-----
11 files changed, 52 insertions(+), 38 deletions(-)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
index 914275eb163..8f4fef293bd 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
@@ -148,7 +148,7 @@ class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable
this.clientBuilder = clientBuilder;
this.client = client;
this.pulsarAdmin = pulsarAdmin;
- this.topicSchema = new TopicSchema(client);
+ this.topicSchema = new TopicSchema(client, Thread.currentThread().getContextClassLoader());
this.statsManager = statsManager;
this.producerBuilder = (ProducerBuilderImpl<?>) client.newProducer().blockIfQueueFull(true).enableBatching(true)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
index cdd662bc75f..a3bc52c98c5 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
@@ -345,7 +345,7 @@ public class PulsarSink<T> implements Sink<T> {
ComponentStatsManager stats, ClassLoader functionClassLoader) {
this.client = client;
this.pulsarSinkConfig = pulsarSinkConfig;
- this.topicSchema = new TopicSchema(client);
+ this.topicSchema = new TopicSchema(client, functionClassLoader);
this.properties = properties;
this.stats = stats;
this.functionClassLoader = functionClassLoader;
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
index 1fb76459e60..18b541ebe59 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
@@ -53,7 +53,7 @@ public abstract class PulsarSource<T> implements Source<T> {
ClassLoader functionClassLoader) {
this.pulsarClient = pulsarClient;
this.pulsarSourceConfig = pulsarSourceConfig;
- this.topicSchema = new TopicSchema(pulsarClient);
+ this.topicSchema = new TopicSchema(pulsarClient, functionClassLoader);
this.properties = properties;
this.functionClassLoader = functionClassLoader;
}
@@ -168,8 +168,8 @@ public abstract class PulsarSource<T> implements Source<T> {
Class<?> typeArg) {
PulsarSourceConsumerConfig.PulsarSourceConsumerConfigBuilder<T> consumerConfBuilder =
PulsarSourceConsumerConfig.<T>builder().isRegexPattern(conf.isRegexPattern())
- .receiverQueueSize(conf.getReceiverQueueSize())
- .consumerProperties(conf.getConsumerProperties());
+ .receiverQueueSize(conf.getReceiverQueueSize())
+ .consumerProperties(conf.getConsumerProperties());
Schema<T> schema;
if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/SingleConsumerPulsarSource.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/SingleConsumerPulsarSource.java
index 885277351b1..041f601ad2e 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/SingleConsumerPulsarSource.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/SingleConsumerPulsarSource.java
@@ -43,14 +43,12 @@ public class SingleConsumerPulsarSource<T> extends PulsarSource<T> {
private Consumer<T> consumer;
private final List<Consumer<T>> inputConsumers = new LinkedList<>();
- public SingleConsumerPulsarSource(PulsarClient pulsarClient,
- SingleConsumerPulsarSourceConfig pulsarSourceConfig,
- Map<String, String> properties,
- ClassLoader functionClassLoader) {
+ public SingleConsumerPulsarSource(PulsarClient pulsarClient, SingleConsumerPulsarSourceConfig pulsarSourceConfig,
+ Map<String, String> properties, ClassLoader functionClassLoader) {
super(pulsarClient, pulsarSourceConfig, properties, functionClassLoader);
this.pulsarClient = pulsarClient;
this.pulsarSourceConfig = pulsarSourceConfig;
- this.topicSchema = new TopicSchema(pulsarClient);
+ this.topicSchema = new TopicSchema(pulsarClient, functionClassLoader);
this.properties = properties;
this.functionClassLoader = functionClassLoader;
}
@@ -59,8 +57,7 @@ public class SingleConsumerPulsarSource<T> extends PulsarSource<T> {
public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
log.info("Opening pulsar source with config: {}", pulsarSourceConfig);
- Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(),
- this.functionClassLoader);
+ Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(), this.functionClassLoader);
checkArgument(!Void.class.equals(typeArg), "Input type of Pulsar Function cannot be Void");
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
index f3a1aaef949..0a4c4058fb4 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/TopicSchema.java
@@ -19,7 +19,11 @@
package org.apache.pulsar.functions.source;
import io.netty.buffer.ByteBuf;
+import java.net.URL;
+import java.net.URLClassLoader;
import java.nio.ByteBuffer;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -49,8 +53,13 @@ public class TopicSchema {
private final Map<String, Schema<?>> cachedSchemas = new HashMap<>();
private final PulsarClient client;
- public TopicSchema(PulsarClient client) {
+ private final ClassLoader functionsClassloader;
+
+ public TopicSchema(PulsarClient client, ClassLoader functionsClassloader) {
this.client = client;
+ this.functionsClassloader = AccessController.doPrivileged(
+ (PrivilegedAction<URLClassLoader>) () -> new URLClassLoader(new URL[0], functionsClassloader)
+ );
}
/**
@@ -244,11 +253,11 @@ public class TopicSchema {
@SuppressWarnings("unchecked")
private <T> Schema<T> newSchemaInstance(String topic, Class<T> clazz, String schemaTypeOrClassName, boolean input) {
return newSchemaInstance(topic, clazz, new ConsumerConfig(schemaTypeOrClassName), input,
- Thread.currentThread().getContextClassLoader());
+ functionsClassloader);
}
@SuppressWarnings("unchecked")
private <T> Schema<T> newSchemaInstance(String topic, Class<T> clazz, ConsumerConfig conf, boolean input) {
- return newSchemaInstance(topic, clazz, conf, input, Thread.currentThread().getContextClassLoader());
+ return newSchemaInstance(topic, clazz, conf, input, functionsClassloader);
}
}
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/TopicSchemaTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/TopicSchemaTest.java
index 26ae6c84893..64de4bcb43f 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/TopicSchemaTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/TopicSchemaTest.java
@@ -18,8 +18,6 @@
*/
package org.apache.pulsar.functions.source;
-import static org.testng.Assert.assertEquals;
-import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.schema.AvroSchema;
@@ -30,12 +28,16 @@ import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.proto.Request;
import org.testng.annotations.Test;
+import java.util.Optional;
+
+import static org.testng.Assert.assertEquals;
+
@Slf4j
public class TopicSchemaTest {
@Test
public void testGetSchema() {
- TopicSchema topicSchema = new TopicSchema(null);
+ TopicSchema topicSchema = new TopicSchema(null, Thread.currentThread().getContextClassLoader());
String TOPIC = "public/default/test";
Schema<?> schema = topicSchema.getSchema(TOPIC + "1", DummyClass.class, Optional.of(SchemaType.JSON));
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
index 908d95784d6..a2ff21c7176 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
@@ -24,10 +24,10 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
-
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.swagger.util.Json;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
@@ -39,8 +39,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-
-import io.swagger.util.Json;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
@@ -72,8 +70,8 @@ import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.apache.pulsar.functions.api.examples.AutoSchemaFunction;
import org.apache.pulsar.functions.api.examples.AvroSchemaTestFunction;
-import org.apache.pulsar.functions.api.examples.MergeTopicFunction;
import org.apache.pulsar.functions.api.examples.InitializableFunction;
+import org.apache.pulsar.functions.api.examples.MergeTopicFunction;
import org.apache.pulsar.functions.api.examples.RecordFunction;
import org.apache.pulsar.functions.api.examples.pojo.AvroTestObject;
import org.apache.pulsar.functions.api.examples.pojo.Users;
@@ -883,6 +881,7 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
String functionName,
String functionFile,
String functionClass,
+ Map<String, String> inputSerdeClassNames,
String outputSerdeClassName,
Map<String, String> userConfigs) throws Exception {
@@ -897,6 +896,7 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
}
generator.setSinkTopic(outputTopicName);
generator.setFunctionName(functionName);
+ generator.setCustomSerDeSourceTopics(inputSerdeClassNames);
generator.setOutputSerDe(outputSerdeClassName);
if (userConfigs != null) {
generator.setUserConfig(userConfigs);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
index e80f811fd5e..0fe77c603b9 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
@@ -54,7 +54,7 @@ public abstract class PulsarFunctionsTestBase extends PulsarTestSuite {
public static final String SERDE_JAVA_CLASS =
"org.apache.pulsar.functions.api.examples.CustomBaseToBaseFunction";
- public static final String SERDE_OUTPUT_CLASS =
+ public static final String SERDE_CLASS =
"org.apache.pulsar.functions.api.examples.CustomBaseSerde";
public static final String EXCLAMATION_PYTHON_CLASS =
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaTest.java
index 7e1c6500e3c..1ba373b64d0 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaTest.java
@@ -19,9 +19,9 @@
package org.apache.pulsar.tests.integration.functions.java;
import static org.testng.Assert.assertEquals;
-
import java.util.Collections;
-
+import java.util.Map;
+import org.apache.commons.collections4.map.HashedMap;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.common.policies.data.FunctionStatus;
import org.apache.pulsar.common.policies.data.FunctionStatusUtil;
@@ -70,10 +70,13 @@ public class PulsarFunctionsJavaTest extends PulsarFunctionsTest {
admin.topics().createNonPartitionedTopic(outputTopicName);
}
+ Map<String, String> inputTopicsSerde = new HashedMap<>();
+ inputTopicsSerde.put(inputTopicName, SERDE_CLASS);
+
String functionName = "test-serde-fn-" + randomName(8);
submitFunction(
- Runtime.JAVA, inputTopicName, outputTopicName, functionName, null, SERDE_JAVA_CLASS,
- SERDE_OUTPUT_CLASS, Collections.singletonMap("serde-topic", outputTopicName)
+ Runtime.JAVA, inputTopicName, outputTopicName, functionName, null, SERDE_JAVA_CLASS, inputTopicsSerde,
+ SERDE_CLASS, Collections.singletonMap("serde-topic", outputTopicName)
);
// get function info
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarWorkerRebalanceDrainTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarWorkerRebalanceDrainTest.java
index 4fd257e1c0c..5a724081b0d 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarWorkerRebalanceDrainTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/java/PulsarWorkerRebalanceDrainTest.java
@@ -18,6 +18,9 @@
*/
package org.apache.pulsar.tests.integration.functions.java;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import com.fasterxml.jackson.databind.MappingIterator;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -26,10 +29,9 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
-import com.fasterxml.jackson.databind.MappingIterator;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
+import org.apache.commons.collections4.map.HashedMap;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.common.functions.WorkerInfo;
import org.apache.pulsar.common.policies.data.FunctionStatus;
@@ -41,8 +43,6 @@ import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator.Runt
import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
import org.testng.annotations.Test;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
@Slf4j
public class PulsarWorkerRebalanceDrainTest extends PulsarFunctionsTest {
@@ -251,9 +251,12 @@ public class PulsarWorkerRebalanceDrainTest extends PulsarFunctionsTest {
admin.topics().createNonPartitionedTopic(outputTopicName);
}
+ Map<String, String> inputTopicsSerde = new HashedMap<>();
+ inputTopicsSerde.put(inputTopicName, SERDE_CLASS);
+
submitFunction(
- Runtime.JAVA, inputTopicName, outputTopicName, functionName, null, SERDE_JAVA_CLASS,
- SERDE_OUTPUT_CLASS, Collections.singletonMap(topicPrefix, outputTopicName)
+ Runtime.JAVA, inputTopicName, outputTopicName, functionName, null, SERDE_JAVA_CLASS, inputTopicsSerde,
+ SERDE_CLASS, Collections.singletonMap(topicPrefix, outputTopicName)
);
}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
index 4edf4624c63..4f3078902ad 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
@@ -46,7 +46,7 @@ public class CommandGenerator {
private String functionClassName;
private String sourceTopic;
private String sourceTopicPattern;
- private Map<String, String> customSereSourceTopics;
+ private Map<String, String> customSerDeSourceTopics;
private String sinkTopic;
private String logTopic;
private String outputSerDe;
@@ -174,8 +174,8 @@ public class CommandGenerator {
if (batchBuilder != null) {
commandBuilder.append("--batch-builder" + batchBuilder);
}
- if (customSereSourceTopics != null && !customSereSourceTopics.isEmpty()) {
- commandBuilder.append(" --customSerdeInputs \'" + new Gson().toJson(customSereSourceTopics) + "\'");
+ if (customSerDeSourceTopics != null && !customSerDeSourceTopics.isEmpty()) {
+ commandBuilder.append(" --customSerdeInputs \'" + new Gson().toJson(customSerDeSourceTopics) + "\'");
}
if (sinkTopic != null) {
commandBuilder.append(" --output " + sinkTopic);
@@ -266,8 +266,8 @@ public class CommandGenerator {
if (StringUtils.isNotEmpty(sourceTopic)) {
commandBuilder.append(" --inputs " + sourceTopic);
}
- if (customSereSourceTopics != null && !customSereSourceTopics.isEmpty()) {
- commandBuilder.append(" --customSerdeInputs \'" + new Gson().toJson(customSereSourceTopics) + "\'");
+ if (customSerDeSourceTopics != null && !customSerDeSourceTopics.isEmpty()) {
+ commandBuilder.append(" --customSerdeInputs \'" + new Gson().toJson(customSerDeSourceTopics) + "\'");
}
if (batchBuilder != null) {
commandBuilder.append("--batch-builder" + batchBuilder);