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 2025/08/18 03:30:29 UTC
(pulsar) branch branch-4.0 updated (332f1fe8271 -> 796603ea1b0)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a change to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 332f1fe8271 [fix][test]fix flaky ZeroQueueSizeTest.testZeroQueueGetExceptionWhenReceiveBatchMessage (#24630)
new eb17634a825 [fix][ws] Fix WebSocket proxy originalPrincipal for HTTP admin API calls (#24613)
new 3aaf265f42c [fix][ws] Fix WebSocket authentication with authenticateOriginalAuthData enabled (#24615)
new 47b73076e13 [improve][broker]Improve the anti-concurrency mechanism expirationCheckInProgress (#24607)
new 4a2dde2614f [improve][broker]Remove block calling that named cursor.asyncGetNth when expiring messages (#24606)
new 36cd2c2ab32 [improve][broker] Upgrade avro version to 1.12.0 (#24617)
new 796603ea1b0 [improve][io] Add dependency file name information to error message when .nar file validation fails with ZipException (#24604)
The 6 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:
distribution/server/src/assemble/LICENSE.bin.txt | 4 +-
distribution/shell/src/assemble/LICENSE.bin.txt | 4 +-
pom.xml | 2 +-
.../pulsar/broker/service/MessageExpirer.java | 3 +
.../apache/pulsar/broker/service/ServerCnx.java | 4 +-
.../nonpersistent/NonPersistentSubscription.java | 6 +
.../persistent/PersistentMessageExpiryMonitor.java | 10 +-
.../service/persistent/PersistentReplicator.java | 17 +++
.../service/persistent/PersistentSubscription.java | 20 +++
.../broker/service/persistent/PersistentTopic.java | 41 +++++-
.../schema/JsonSchemaCompatibilityCheck.java | 3 +-
.../service/PersistentMessageFinderTest.java | 141 ++++++++++++++++++---
.../pulsar/broker/service/ReplicatorTest.java | 30 +++++
.../PersistentMessageExpiryMonitorTest.java | 141 +++++++++++++++++++++
.../schema/validator/SchemaDataValidatorTest.java | 16 ++-
.../PulsarFunctionTestTemporaryDirectory.java | 4 +
.../apache/pulsar/io/AbstractPulsarE2ETest.java | 3 +-
.../apache/pulsar/io/PulsarFunctionE2ETest.java | 23 ++++
...IntegrationTest.java => ProxyRoleAuthTest.java} | 27 ++--
...er.java => ProxyRoleAuthWebServiceURLTest.java} | 27 ++--
...ProxyRoleAuthenticateOriginalAuthDataTest.java} | 14 +-
...enticateOriginalAuthDataWebServiceURLTest.java} | 15 ++-
.../org/apache/pulsar/client/impl/HttpClient.java | 8 ++
.../pulsar/client/impl/schema/util/SchemaUtil.java | 3 +-
.../pulsar/client/impl/schema/AvroSchemaTest.java | 17 ---
.../functions/utils/FunctionFilePackage.java | 3 +-
.../pulsar/io/elasticsearch/JsonConverter.java | 4 +-
.../pulsar/io/kinesis/json/JsonConverter.java | 4 +-
28 files changed, 507 insertions(+), 87 deletions(-)
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java
rename pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/{WebSocketProxyAuthIntegrationTest.java => ProxyRoleAuthTest.java} (98%)
copy pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/{MockUnauthenticationProvider.java => ProxyRoleAuthWebServiceURLTest.java} (52%)
copy pulsar-broker/src/test/java/org/apache/pulsar/{broker/service/schema/ProtobufSchemaCompatibilityCheckTest.java => websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataTest.java} (70%)
copy pulsar-broker/src/test/java/org/apache/pulsar/{broker/service/schema/AvroSchemaCompatibilityCheckTest.java => websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataWebServiceURLTest.java} (67%)
(pulsar) 01/06: [fix][ws] Fix WebSocket proxy originalPrincipal for HTTP admin API calls (#24613)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit eb17634a825079b92aa98b2cc3aa96f77e94d7db
Author: Penghui Li <pe...@apache.org>
AuthorDate: Fri Aug 8 10:31:42 2025 -0700
[fix][ws] Fix WebSocket proxy originalPrincipal for HTTP admin API calls (#24613)
Co-authored-by: Claude <no...@anthropic.com>
---
...IntegrationTest.java => ProxyRoleAuthTest.java} | 27 ++++++++-------
.../proxy/ProxyRoleAuthWebServiceURLTest.java | 38 ++++++++++++++++++++++
.../org/apache/pulsar/client/impl/HttpClient.java | 8 +++++
3 files changed, 62 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/WebSocketProxyAuthIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthTest.java
similarity index 98%
rename from pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/WebSocketProxyAuthIntegrationTest.java
rename to pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthTest.java
index 624c2522822..595991c40aa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/WebSocketProxyAuthIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthTest.java
@@ -68,9 +68,9 @@ import org.testng.annotations.Test;
* 2. testWebSocketProxyWithUnauthorizedToken: Negative test with unauthorized tokens
*/
@Test(groups = "websocket")
-public class WebSocketProxyAuthIntegrationTest extends ProducerConsumerBase {
+public class ProxyRoleAuthTest extends ProducerConsumerBase {
- private static final Logger log = LoggerFactory.getLogger(WebSocketProxyAuthIntegrationTest.class);
+ private static final Logger log = LoggerFactory.getLogger(ProxyRoleAuthTest.class);
// JWT token authentication setup with different roles
private static final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
@@ -129,6 +129,19 @@ public class WebSocketProxyAuthIntegrationTest extends ProducerConsumerBase {
// Setup namespace and grant permissions for client role
setupNamespacePermissions();
+ WebSocketProxyConfiguration proxyConfig = getProxyConfig();
+
+ service = spyWithClassAndConstructorArgs(WebSocketService.class, proxyConfig);
+ doReturn(registerCloseable(new ZKMetadataStore(mockZooKeeperGlobal))).when(service)
+ .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
+
+ proxyServer = new ProxyServer(proxyConfig);
+ WebSocketServiceStarter.start(proxyServer, service);
+
+ log.info("WebSocket Proxy Server started on port: {}", proxyServer.getListenPortHTTP().get());
+ }
+
+ protected WebSocketProxyConfiguration getProxyConfig() {
// Create WebSocket proxy configuration with authentication and authorization enabled
WebSocketProxyConfiguration proxyConfig = new WebSocketProxyConfiguration();
proxyConfig.setWebServicePort(Optional.of(0));
@@ -152,15 +165,7 @@ public class WebSocketProxyAuthIntegrationTest extends ProducerConsumerBase {
// Set broker service URL to connect to our test broker
proxyConfig.setBrokerServiceUrl(pulsar.getBrokerServiceUrl());
proxyConfig.setBrokerServiceUrlTls(pulsar.getBrokerServiceUrlTls());
-
- service = spyWithClassAndConstructorArgs(WebSocketService.class, proxyConfig);
- doReturn(registerCloseable(new ZKMetadataStore(mockZooKeeperGlobal))).when(service)
- .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
-
- proxyServer = new ProxyServer(proxyConfig);
- WebSocketServiceStarter.start(proxyServer, service);
-
- log.info("WebSocket Proxy Server started on port: {}", proxyServer.getListenPortHTTP().get());
+ return proxyConfig;
}
@AfterMethod(alwaysRun = true)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
new file mode 100644
index 00000000000..f726178aabc
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
@@ -0,0 +1,38 @@
+/*
+ * 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.websocket.proxy;
+
+import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+
+/**
+ * Same test with ProxyRoleAuthTest but using REST API as the internal client.
+ */
+public class ProxyRoleAuthWebServiceURLTest extends ProxyRoleAuthTest {
+
+ @Override
+ protected WebSocketProxyConfiguration getProxyConfig() {
+ // Create WebSocket proxy configuration with authentication and authorization enabled
+ WebSocketProxyConfiguration proxyConfig = super.getProxyConfig();
+ proxyConfig.setServiceUrl(pulsar.getWebServiceAddress());
+ proxyConfig.setServiceUrlTls(pulsar.getWebServiceAddressTls());
+ proxyConfig.setBrokerServiceUrl(null);
+ proxyConfig.setBrokerServiceUrlTls(null);
+ return proxyConfig;
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
index 8e448d801fa..d0739ce855f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
@@ -59,18 +59,21 @@ import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
@Slf4j
public class HttpClient implements Closeable {
+ private static final String ORIGINAL_PRINCIPAL_HEADER = "X-Original-Principal";
protected static final int DEFAULT_CONNECT_TIMEOUT_IN_SECONDS = 10;
protected static final int DEFAULT_READ_TIMEOUT_IN_SECONDS = 30;
protected final AsyncHttpClient httpClient;
protected final ServiceNameResolver serviceNameResolver;
protected final Authentication authentication;
+ protected final ClientConfigurationData clientConf;
protected ScheduledExecutorService executorService;
protected PulsarSslFactory sslFactory;
protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException {
this.authentication = conf.getAuthentication();
this.serviceNameResolver = new PulsarServiceNameResolver();
+ this.clientConf = conf;
this.serviceNameResolver.updateServiceUrl(conf.getServiceUrl());
DefaultAsyncHttpClientConfig.Builder confBuilder = new DefaultAsyncHttpClientConfig.Builder();
@@ -192,6 +195,11 @@ public class HttpClient implements Closeable {
}
}
+ // Add X-Original-Principal header if originalPrincipal is configured (for proxy scenarios)
+ if (clientConf.getOriginalPrincipal() != null) {
+ builder.addHeader(ORIGINAL_PRINCIPAL_HEADER, clientConf.getOriginalPrincipal());
+ }
+
builder.execute().toCompletableFuture().whenComplete((response2, t) -> {
if (t != null) {
log.warn("[{}] Failed to perform http request: {}", requestUrl, t.getMessage());
(pulsar) 04/06: [improve][broker]Remove block calling that named cursor.asyncGetNth when expiring messages (#24606)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 4a2dde2614f21a1ab5effc5ecec601e3feea447b
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Aug 8 12:46:17 2025 +0800
[improve][broker]Remove block calling that named cursor.asyncGetNth when expiring messages (#24606)
---
.../pulsar/broker/service/MessageExpirer.java | 3 +
.../nonpersistent/NonPersistentSubscription.java | 6 ++
.../persistent/PersistentMessageExpiryMonitor.java | 6 ++
.../service/persistent/PersistentReplicator.java | 17 ++++
.../service/persistent/PersistentSubscription.java | 20 ++++
.../broker/service/persistent/PersistentTopic.java | 41 +++++++-
.../service/PersistentMessageFinderTest.java | 111 +++++++++++++++++++++
.../pulsar/broker/service/ReplicatorTest.java | 30 ++++++
8 files changed, 231 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/MessageExpirer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/MessageExpirer.java
index 7cb1d2a904a..3008717a3df 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/MessageExpirer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/MessageExpirer.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.broker.service;
+import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.Position;
import org.apache.pulsar.common.classification.InterfaceStability;
@@ -27,4 +28,6 @@ public interface MessageExpirer {
boolean expireMessages(Position position);
boolean expireMessages(int messageTTLInSeconds);
+
+ CompletableFuture<Boolean> expireMessagesAsync(int messageTTLInSeconds);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
index d469ce1daa1..549a17b2ae4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
@@ -472,6 +472,12 @@ public class NonPersistentSubscription extends AbstractSubscription {
+ " non-persistent topic.");
}
+ @Override
+ public CompletableFuture<Boolean> expireMessagesAsync(int messageTTLInSeconds) {
+ return CompletableFuture.failedFuture(new UnsupportedOperationException("Expire message by timestamp is not"
+ + " supported for non-persistent topic."));
+ }
+
@Override
public boolean expireMessages(Position position) {
throw new UnsupportedOperationException("Expire message by position is not supported for"
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
index a5ee05db767..348284fc624 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -22,6 +22,7 @@ import com.google.common.annotations.VisibleForTesting;
import java.util.Objects;
import java.util.Optional;
import java.util.SortedMap;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
import org.apache.bookkeeper.mledger.AsyncCallbacks.FindEntryCallback;
@@ -79,6 +80,11 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback, Messag
&& this.cursor.getManagedLedger().getConfig().isAutoSkipNonRecoverableData();
}
+ @Override
+ public CompletableFuture<Boolean> expireMessagesAsync(int messageTTLInSeconds) {
+ return CompletableFuture.supplyAsync(() -> expireMessages(messageTTLInSeconds), topic.getOrderedExecutor());
+ }
+
@Override
public boolean expireMessages(int messageTTLInSeconds) {
if (expirationCheckInProgressUpdater.compareAndSet(this, FALSE, TRUE)) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index 5952bb14d22..559b39e0fb4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -681,6 +681,23 @@ public abstract class PersistentReplicator extends AbstractReplicator
return expiryMonitor.expireMessages(messageTTLInSeconds);
}
+ @Override
+ public CompletableFuture<Boolean> expireMessagesAsync(int messageTTLInSeconds) {
+ long backlog = cursor.getNumberOfEntriesInBacklog(false);
+ if (backlog == 0) {
+ return CompletableFuture.completedFuture(false);
+ } else if (backlog < MINIMUM_BACKLOG_FOR_EXPIRY_CHECK) {
+ return topic.isOldestMessageExpiredAsync(cursor, messageTTLInSeconds).thenCompose(oldestMsgExpired -> {
+ if (oldestMsgExpired) {
+ return expiryMonitor.expireMessagesAsync(messageTTLInSeconds);
+ } else {
+ return CompletableFuture.completedFuture(false);
+ }
+ });
+ }
+ return expiryMonitor.expireMessagesAsync(messageTTLInSeconds);
+ }
+
@Override
public boolean expireMessages(Position position) {
return expiryMonitor.expireMessages(position);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index 5a9c171183c..371ee1bcc76 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -1232,6 +1232,26 @@ public class PersistentSubscription extends AbstractSubscription {
return expiryMonitor.expireMessages(messageTTLInSeconds);
}
+ @Override
+ public CompletableFuture<Boolean> expireMessagesAsync(int messageTTLInSeconds) {
+ long backlog = getNumberOfEntriesInBacklog(false);
+ if (backlog == 0) {
+ return CompletableFuture.completedFuture(false);
+ }
+ if (dispatcher != null && dispatcher.isConsumerConnected() && backlog < MINIMUM_BACKLOG_FOR_EXPIRY_CHECK) {
+ return topic.isOldestMessageExpiredAsync(cursor, messageTTLInSeconds)
+ .thenCompose(oldestMsgExpired -> {
+ if (oldestMsgExpired) {
+ this.lastExpireTimestamp = System.currentTimeMillis();
+ return expiryMonitor.expireMessagesAsync(messageTTLInSeconds);
+ } else {
+ return CompletableFuture.completedFuture(false);
+ }
+ });
+ }
+ return expiryMonitor.expireMessagesAsync(messageTTLInSeconds);
+ }
+
@Override
public boolean expireMessages(Position position) {
this.lastExpireTimestamp = System.currentTimeMillis();
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 e59bcd94263..14f73ce8a76 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
@@ -31,6 +31,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import io.netty.buffer.ByteBuf;
import io.netty.util.concurrent.FastThreadLocal;
+import java.io.IOException;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Collections;
@@ -2091,13 +2092,13 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
if (!isCompactionSubscription(sub.getName())
&& (additionalSystemCursorNames.isEmpty()
|| !additionalSystemCursorNames.contains(sub.getName()))) {
- sub.expireMessages(messageTtlInSeconds);
+ sub.expireMessagesAsync(messageTtlInSeconds);
}
});
replicators.forEach((__, replicator)
- -> ((PersistentReplicator) replicator).expireMessages(messageTtlInSeconds));
+ -> ((PersistentReplicator) replicator).expireMessagesAsync(messageTtlInSeconds));
shadowReplicators.forEach((__, replicator)
- -> ((PersistentReplicator) replicator).expireMessages(messageTtlInSeconds));
+ -> ((PersistentReplicator) replicator).expireMessagesAsync(messageTtlInSeconds));
}
}
@@ -3860,6 +3861,40 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return isOldestMessageExpired;
}
+ public CompletableFuture<Boolean> isOldestMessageExpiredAsync(ManagedCursor cursor, int messageTTLInSeconds) {
+ CompletableFuture<Boolean> res = new CompletableFuture<>();
+ cursor.asyncGetNthEntry(1, IndividualDeletedEntries.Include, new AsyncCallbacks.ReadEntryCallback() {
+
+ @Override
+ public void readEntryComplete(Entry entry, Object ctx) {
+ long entryTimestamp = 0;
+ try {
+ entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ res.complete(MessageImpl.isEntryExpired(
+ (int) (messageTTLInSeconds * MESSAGE_EXPIRY_THRESHOLD), entryTimestamp));
+ } catch (IOException e) {
+ log.warn("[{}] [{}] Error while getting the oldest message", topic, cursor.toString(), e);
+ res.complete(false);
+ }
+
+ }
+
+ @Override
+ public void readEntryFailed(ManagedLedgerException e, Object ctx) {
+ if (brokerService.pulsar().getConfiguration().isAutoSkipNonRecoverableData()
+ && e instanceof NonRecoverableLedgerException) {
+ // NonRecoverableLedgerException means the ledger or entry can't be read anymore.
+ // if AutoSkipNonRecoverableData is set to true, just return true here.
+ res.complete(true);
+ } else {
+ log.warn("[{}] [{}] Error while getting the oldest message", topic, cursor.toString(), e);
+ res.complete(false);
+ }
+ }
+ }, null);
+ return res;
+ }
+
/**
* Clears backlog for all cursors in the topic.
*
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
index 5a995c63df5..4a7da569ba7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
@@ -59,6 +60,8 @@ import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.Ledge
import org.apache.bookkeeper.test.MockedBookKeeperTestCase;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.service.persistent.PersistentMessageExpiryMonitor;
import org.apache.pulsar.broker.service.persistent.PersistentMessageFinder;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
@@ -450,6 +453,62 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
}
+ /**
+ * It tests that message expiry doesn't get stuck if it can't read deleted ledger's entry.
+ */
+ @Test
+ void testMessageExpiryAsyncWithTimestampNonRecoverableException() throws Exception {
+
+ final String ledgerAndCursorName = "testPersistentMessageExpiryWithNonRecoverableLedgers";
+ final int entriesPerLedger = 2;
+ final int totalEntries = 10;
+ final int ttlSeconds = 1;
+
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setRetentionSizeInMB(10);
+ config.setMaxEntriesPerLedger(entriesPerLedger);
+ config.setRetentionTime(1, TimeUnit.HOURS);
+ config.setAutoSkipNonRecoverableData(true);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerAndCursorName, config);
+ ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
+
+ for (int i = 0; i < totalEntries; i++) {
+ ledger.addEntry(createMessageWrittenToLedger("msg" + i));
+ }
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(ledger.getState(), ManagedLedgerImpl.State.LedgerOpened));
+
+ List<LedgerInfo> ledgers = ledger.getLedgersInfoAsList();
+ LedgerInfo lastLedgerInfo = ledgers.get(ledgers.size() - 1);
+ // The `lastLedgerInfo` should be newly opened, and it does not contain any entries.
+ // Please refer to: https://github.com/apache/pulsar/pull/22034
+ assertEquals(lastLedgerInfo.getEntries(), 0);
+ assertEquals(ledgers.size(), totalEntries / entriesPerLedger + 1);
+
+ // this will make sure that all entries should be deleted
+ Thread.sleep(TimeUnit.SECONDS.toMillis(ttlSeconds));
+
+ bkc.deleteLedger(ledgers.get(0).getLedgerId());
+ bkc.deleteLedger(ledgers.get(1).getLedgerId());
+ bkc.deleteLedger(ledgers.get(2).getLedgerId());
+
+ PersistentTopic mock = mockPersistentTopic("topicname");
+
+ PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(mock, c1.getName(), c1, null);
+ assertTrue(monitor.expireMessagesAsync(ttlSeconds).get());
+ Awaitility.await().untilAsserted(() -> {
+ Position markDeletePosition = c1.getMarkDeletedPosition();
+ // The markDeletePosition points to the last entry of the previous ledger in lastLedgerInfo.
+ assertEquals(markDeletePosition.getLedgerId(), lastLedgerInfo.getLedgerId() - 1);
+ assertEquals(markDeletePosition.getEntryId(), entriesPerLedger - 1);
+ });
+
+ c1.close();
+ ledger.close();
+ factory.shutdown();
+
+ }
+
public void testFindMessageWithTimestampAutoSkipNonRecoverable() throws Exception {
final String ledgerAndCursorName = "testFindMessageWithTimestampAutoSkipNonRecoverable";
@@ -625,6 +684,21 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
assertEquals(c1.getNumberOfEntriesInBacklog(true), 0);
}
+
+ private PersistentTopic mockPersistentTopic(String topicName) throws Exception {
+ PersistentTopic mock = mock(PersistentTopic.class);
+ when(mock.getName()).thenReturn("topicname");
+ when(mock.getLastPosition()).thenReturn(PositionFactory.EARLIEST);
+ BrokerService brokerService = mock(BrokerService.class);
+ doReturn(brokerService).when(mock).getBrokerService();
+ doReturn(executor).when(mock).getOrderedExecutor();
+ PulsarService pulsarService = mock(PulsarService.class);
+ doReturn(pulsarService).when(brokerService).pulsar();
+ ServiceConfiguration serviceConfiguration = new ServiceConfiguration();
+ doReturn(serviceConfiguration).when(pulsarService).getConfig();
+ return mock;
+ }
+
@Test
public void testCheckExpiryByLedgerClosureTimeWithAckUnclosedLedger() throws Throwable {
final String ledgerAndCursorName = "testCheckExpiryByLedgerClosureTimeWithAckUnclosedLedger";
@@ -664,6 +738,43 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
Assert.assertNull(throwableAtomicReference.get());
}
+ @Test
+ public void testCheckExpiryAsyncByLedgerClosureTimeWithAckUnclosedLedger() throws Throwable {
+ final String ledgerAndCursorName = "testCheckExpiryByLedgerClosureTimeWithAckUnclosedLedger";
+ int maxTTLSeconds = 1;
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(5);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(ledgerAndCursorName, config);
+ ManagedCursorImpl c1 = (ManagedCursorImpl) ledger.openCursor(ledgerAndCursorName);
+ // set client clock to 10 days later
+ long incorrectPublishTimestamp = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(10);
+ for (int i = 0; i < 7; i++) {
+ ledger.addEntry(createMessageWrittenToLedger("msg" + i, incorrectPublishTimestamp));
+ }
+ assertEquals(ledger.getLedgersInfoAsList().size(), 2);
+ PersistentTopic mock = mockPersistentTopic("topicname");
+ PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(mock, c1.getName(), c1, null);
+ AsyncCallbacks.MarkDeleteCallback markDeleteCallback =
+ (AsyncCallbacks.MarkDeleteCallback) spy(
+ FieldUtils.readDeclaredField(monitor, "markDeleteCallback", true));
+ FieldUtils.writeField(monitor, "markDeleteCallback", markDeleteCallback, true);
+
+ AtomicReference<Throwable> throwableAtomicReference = new AtomicReference<>();
+ Mockito.doAnswer(invocation -> {
+ ManagedLedgerException argument = invocation.getArgument(0, ManagedLedgerException.class);
+ throwableAtomicReference.set(argument);
+ return invocation.callRealMethod();
+ }).when(markDeleteCallback).markDeleteFailed(any(), any());
+
+ Position position = ledger.getLastConfirmedEntry();
+ c1.markDelete(position);
+ Thread.sleep(TimeUnit.SECONDS.toMillis(maxTTLSeconds));
+ monitor.expireMessagesAsync(maxTTLSeconds).get();
+ assertEquals(c1.getNumberOfEntriesInBacklog(true), 0);
+
+ Assert.assertNull(throwableAtomicReference.get());
+ }
+
@Test
void testMessageExpiryWithPosition() throws Exception {
final String ledgerAndCursorName = "testPersistentMessageExpiryWithPositionNonRecoverableLedgers";
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 a252de934af..2e1ed43bd8a 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
@@ -669,6 +669,36 @@ public class ReplicatorTest extends ReplicatorTestBase {
assertEquals(status.getReplicationBacklog(), 0);
}
+ @Test(timeOut = 30000)
+ public void testReplicatorExpireMsgAsync() throws Exception {
+
+ // This test is to verify that reset cursor fails on global topic
+ SortedSet<String> testDests = new TreeSet<>();
+
+ final TopicName dest = TopicName
+ .get(BrokerTestUtil.newUniqueName("persistent://pulsar/ns/clearBacklogTopic"));
+ testDests.add(dest.toString());
+
+ @Cleanup
+ MessageProducer producer1 = new MessageProducer(url1, dest);
+
+ @Cleanup
+ MessageConsumer consumer1 = new MessageConsumer(url3, dest);
+
+ // Produce from cluster1 and consume from the rest
+ producer1.produce(2);
+ PersistentTopic topic = (PersistentTopic) pulsar1.getBrokerService().getTopicReference(dest.toString()).get();
+ PersistentReplicator replicator = (PersistentReplicator) spy(
+ topic.getReplicators().get(topic.getReplicators().keySet().stream().toList().get(0)));
+ replicator.readEntriesFailed(new ManagedLedgerException.InvalidCursorPositionException("failed"), null);
+ replicator.clearBacklog().get();
+ Thread.sleep(100);
+ replicator.updateRates(); // for code-coverage
+ replicator.expireMessagesAsync(1).get(); // for code-coverage
+ ReplicatorStats status = replicator.computeStats();
+ assertEquals(status.getReplicationBacklog(), 0);
+ }
+
@Test(timeOut = 30000)
public void testResetReplicatorSubscriptionPosition() throws Exception {
(pulsar) 02/06: [fix][ws] Fix WebSocket authentication with authenticateOriginalAuthData enabled (#24615)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 3aaf265f42cbae44b25444744cd7181ce21d53c4
Author: Penghui Li <pe...@apache.org>
AuthorDate: Sun Aug 10 20:21:49 2025 -0700
[fix][ws] Fix WebSocket authentication with authenticateOriginalAuthData enabled (#24615)
---
.../org/apache/pulsar/broker/service/ServerCnx.java | 4 +++-
.../proxy/ProxyRoleAuthWebServiceURLTest.java | 2 ++
... => ProxyRoleAuthenticateOriginalAuthDataTest.java} | 16 ++++++----------
...AuthenticateOriginalAuthDataWebServiceURLTest.java} | 18 +++++++-----------
4 files changed, 18 insertions(+), 22 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 71cbefdcb5c..67b4b082343 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -28,6 +28,7 @@ import static org.apache.pulsar.broker.lookup.TopicLookupBase.lookupTopicAsync;
import static org.apache.pulsar.broker.service.persistent.PersistentTopic.getMigratedClusterUrl;
import static org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.ignoreUnrecoverableBKException;
import static org.apache.pulsar.common.api.proto.ProtocolVersion.v5;
+import static org.apache.pulsar.common.naming.Constants.WEBSOCKET_DUMMY_ORIGINAL_PRINCIPLE;
import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
import static org.apache.pulsar.common.protocol.Commands.newLookupErrorResponse;
import com.google.common.annotations.VisibleForTesting;
@@ -1130,7 +1131,8 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
log.debug("[{}] Authenticate role : {}", remoteAddress, role);
}
- if (connect.hasOriginalPrincipal() && service.getPulsar().getConfig().isAuthenticateOriginalAuthData()) {
+ if (connect.hasOriginalPrincipal() && service.getPulsar().getConfig().isAuthenticateOriginalAuthData()
+ && !WEBSOCKET_DUMMY_ORIGINAL_PRINCIPLE.equals(connect.getOriginalPrincipal())) {
// Flow:
// 1. Initialize original authentication.
// 2. Authenticate the proxy's authentication data.
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
index f726178aabc..08087f6f3e0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
@@ -19,10 +19,12 @@
package org.apache.pulsar.websocket.proxy;
import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+import org.testng.annotations.Test;
/**
* Same test with ProxyRoleAuthTest but using REST API as the internal client.
*/
+@Test(groups = "websocket")
public class ProxyRoleAuthWebServiceURLTest extends ProxyRoleAuthTest {
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataTest.java
similarity index 60%
copy from pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
copy to pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataTest.java
index f726178aabc..0f292669117 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataTest.java
@@ -18,21 +18,17 @@
*/
package org.apache.pulsar.websocket.proxy;
-import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+import org.testng.annotations.Test;
/**
* Same test with ProxyRoleAuthTest but using REST API as the internal client.
*/
-public class ProxyRoleAuthWebServiceURLTest extends ProxyRoleAuthTest {
+@Test(groups = "websocket")
+public class ProxyRoleAuthenticateOriginalAuthDataTest extends ProxyRoleAuthTest {
@Override
- protected WebSocketProxyConfiguration getProxyConfig() {
- // Create WebSocket proxy configuration with authentication and authorization enabled
- WebSocketProxyConfiguration proxyConfig = super.getProxyConfig();
- proxyConfig.setServiceUrl(pulsar.getWebServiceAddress());
- proxyConfig.setServiceUrlTls(pulsar.getWebServiceAddressTls());
- proxyConfig.setBrokerServiceUrl(null);
- proxyConfig.setBrokerServiceUrlTls(null);
- return proxyConfig;
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ conf.setAuthenticateOriginalAuthData(true);
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataWebServiceURLTest.java
similarity index 55%
copy from pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
copy to pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataWebServiceURLTest.java
index f726178aabc..4078f67fc2b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthWebServiceURLTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyRoleAuthenticateOriginalAuthDataWebServiceURLTest.java
@@ -18,21 +18,17 @@
*/
package org.apache.pulsar.websocket.proxy;
-import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+import org.testng.annotations.Test;
/**
- * Same test with ProxyRoleAuthTest but using REST API as the internal client.
+ * Same test with ProxyRoleAuthWebServiceURLTest but with authenticateOriginalAuthData enabled.
*/
-public class ProxyRoleAuthWebServiceURLTest extends ProxyRoleAuthTest {
+@Test(groups = "websocket")
+public class ProxyRoleAuthenticateOriginalAuthDataWebServiceURLTest extends ProxyRoleAuthWebServiceURLTest {
@Override
- protected WebSocketProxyConfiguration getProxyConfig() {
- // Create WebSocket proxy configuration with authentication and authorization enabled
- WebSocketProxyConfiguration proxyConfig = super.getProxyConfig();
- proxyConfig.setServiceUrl(pulsar.getWebServiceAddress());
- proxyConfig.setServiceUrlTls(pulsar.getWebServiceAddressTls());
- proxyConfig.setBrokerServiceUrl(null);
- proxyConfig.setBrokerServiceUrlTls(null);
- return proxyConfig;
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ conf.setAuthenticateOriginalAuthData(true);
}
}
(pulsar) 03/06: [improve][broker]Improve the anti-concurrency mechanism expirationCheckInProgress (#24607)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 47b73076e13ee5a8b249dccb1362ccee2c6d9b0d
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Thu Aug 7 19:25:24 2025 +0800
[improve][broker]Improve the anti-concurrency mechanism expirationCheckInProgress (#24607)
---
.../persistent/PersistentMessageExpiryMonitor.java | 4 +-
.../service/PersistentMessageFinderTest.java | 30 ++---
.../PersistentMessageExpiryMonitorTest.java | 141 +++++++++++++++++++++
3 files changed, 158 insertions(+), 17 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
index f6ba8a2f9a2..a5ee05db767 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
@@ -194,6 +194,7 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback, Messag
if (subscription != null && subscription.getType() == SubType.Key_Shared) {
subscription.getDispatcher().markDeletePositionMoveForward();
}
+ expirationCheckInProgress = FALSE;
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Mark deleted {} messages", topicName, subName, numMessagesExpired);
}
@@ -202,6 +203,7 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback, Messag
@Override
public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
log.warn("[{}][{}] Message expiry failed - mark delete failed", topicName, subName, exception);
+ expirationCheckInProgress = FALSE;
updateRates();
}
};
@@ -220,9 +222,9 @@ public class PersistentMessageExpiryMonitor implements FindEntryCallback, Messag
if (log.isDebugEnabled()) {
log.debug("[{}][{}] No messages to expire", topicName, subName);
}
+ expirationCheckInProgress = FALSE;
updateRates();
}
- expirationCheckInProgress = FALSE;
}
@Override
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
index 4e0eee7f0b7..5a995c63df5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
@@ -19,12 +19,9 @@
package org.apache.pulsar.broker.service;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -43,6 +40,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.ws.rs.client.Entity;
import javax.ws.rs.core.MediaType;
@@ -691,18 +689,24 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
}
when(topic.getLastPosition()).thenReturn(positions.get(positions.size() - 1));
- PersistentMessageExpiryMonitor monitor = spy(new PersistentMessageExpiryMonitor(topic,
- cursor.getName(), cursor, subscription));
+ AtomicInteger counterCalledFindEntryComplete = new AtomicInteger(0);
+ PersistentMessageExpiryMonitor monitor = new PersistentMessageExpiryMonitor(topic,
+ cursor.getName(), cursor, subscription) {
+ @Override
+ public void findEntryComplete(Position position, Object ctx) {
+ super.findEntryComplete(position, ctx);
+ counterCalledFindEntryComplete.incrementAndGet();
+ }
+ };
assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(0).getLedgerId(), -1));
boolean issued;
// Expire by position and verify mark delete position of cursor.
issued = monitor.expireMessages(positions.get(15));
- Awaitility.await().untilAsserted(() -> verify(monitor, times(1)).findEntryComplete(any(), any()));
+ Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 1));
assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(15).getLedgerId(),
positions.get(15).getEntryId()));
assertTrue(issued);
- clearInvocations(monitor);
// Expire by position beyond last position and nothing should happen.
issued = monitor.expireMessages(PositionFactory.create(100, 100));
@@ -712,32 +716,26 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
// Expire by position again and verify mark delete position of cursor didn't change.
issued = monitor.expireMessages(positions.get(15));
- Awaitility.await().untilAsserted(() -> verify(monitor,
- times(1)).findEntryComplete(any(), any()));
+ Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 2));
assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(15).getLedgerId(),
positions.get(15).getEntryId()));
assertTrue(issued);
- clearInvocations(monitor);
// Expire by position before current mark delete position and
// verify mark delete position of cursor didn't change.
issued = monitor.expireMessages(positions.get(10));
- Awaitility.await().untilAsserted(() -> verify(monitor,
- times(1)).findEntryComplete(any(), any()));
+ Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 3));
assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(15).getLedgerId(),
positions.get(15).getEntryId()));
assertTrue(issued);
- clearInvocations(monitor);
// Expire by position after current mark delete position and
// verify mark delete position of cursor move to new position.
issued = monitor.expireMessages(positions.get(16));
- Awaitility.await().untilAsserted(() -> verify(monitor,
- times(1)).findEntryComplete(any(), any()));
+ Awaitility.await().untilAsserted(() -> assertEquals(counterCalledFindEntryComplete.get(), 4));
assertEquals(cursor.getMarkDeletedPosition(), PositionFactory.create(positions.get(16).getLedgerId(),
positions.get(16).getEntryId()));
assertTrue(issued);
- clearInvocations(monitor);
ManagedCursorImpl mockCursor = mock(ManagedCursorImpl.class);
PersistentMessageExpiryMonitor mockMonitor = spy(new PersistentMessageExpiryMonitor(topic,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java
new file mode 100644
index 00000000000..5535561a5fa
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitorTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.broker.service.persistent;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.testng.AssertJUnit.assertEquals;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class PersistentMessageExpiryMonitorTest extends ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ /***
+ * Confirm the anti-concurrency mechanism "expirationCheckInProgressUpdater" works.
+ */
+ @Test
+ void testConcurrentlyExpireMessages() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String cursorName = "s1";
+ admin.topics().createNonPartitionedTopic(topicName);
+ admin.topics().createSubscriptionAsync(topicName, cursorName, MessageId.earliest);
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
+ producer.send("1");
+ producer.send("2");
+ producer.send("3");
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopic(topicName, false).join().get();
+ ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();
+ ManagedCursorImpl cursor = (ManagedCursorImpl) ml.getCursors().get(cursorName);
+ ManagedCursorImpl spyCursor = Mockito.spy(cursor);
+
+ // Make the mark-deleting delay.
+ CountDownLatch firstFindingCompleted = new CountDownLatch(1);
+ AtomicInteger calledFindPositionCount = new AtomicInteger();
+ doAnswer(invocationOnMock -> {
+ firstFindingCompleted.countDown();
+ ml.getExecutor().execute(() -> {
+ try {
+ Thread.sleep(3000);
+ invocationOnMock.callRealMethod();
+ } catch (Throwable ex) {
+ log.error("Unexpected exception when calling mark delete", ex);
+ }
+ });
+ return true;
+ }).when(spyCursor).asyncMarkDelete(any(Position.class), any(Map.class),
+ any(AsyncCallbacks.MarkDeleteCallback.class), any());
+ doAnswer(invocationOnMock -> {
+ calledFindPositionCount.incrementAndGet();
+ return invocationOnMock.callRealMethod();
+ }).when(spyCursor).asyncFindNewestMatching(any(), any(), any(), any(), any(), any(), anyBoolean());
+ doAnswer(invocationOnMock -> {
+ calledFindPositionCount.incrementAndGet();
+ return invocationOnMock.callRealMethod();
+ }).when(spyCursor).asyncFindNewestMatching(any(), any(), any(), any(), anyBoolean());
+ doAnswer(invocationOnMock -> {
+ calledFindPositionCount.incrementAndGet();
+ return invocationOnMock.callRealMethod();
+ }).when(spyCursor).asyncFindNewestMatching(any(), any(), any(), any());
+
+ // Sleep 2s to make "find(1s)" get a position.
+ Thread.sleep(2000);
+
+ // Start two expire tasks concurrently.
+ PersistentMessageExpiryMonitor
+ monitor = new PersistentMessageExpiryMonitor(persistentTopic, cursorName, spyCursor, null);
+ CompletableFuture<Boolean> expireTask1 = new CompletableFuture<>();
+ new Thread(() -> {
+ expireTask1.complete(monitor.expireMessages(1));
+ }).start();
+ CompletableFuture<Boolean> expireTask2 = new CompletableFuture<>();
+ new Thread(() -> {
+ expireTask2.complete(monitor.expireMessages(1));
+ }).start();
+ firstFindingCompleted.await();
+ CompletableFuture<Boolean> expireTask3 = new CompletableFuture<>();
+ new Thread(() -> {
+ expireTask3.complete(monitor.expireMessages(1));
+ }).start();
+
+ Awaitility.await().untilAsserted(() -> {
+ Assert.assertEquals(monitor.getTotalMessageExpired(), 3);
+ });
+ // Verify: since the other 2 tasks have been prevented, the count of calling find position is 1.
+ Thread.sleep(1000);
+ assertEquals(1, calledFindPositionCount.get());
+
+ // cleanup.
+ producer.close();
+ admin.topics().delete(topicName);
+ }
+}
(pulsar) 05/06: [improve][broker] Upgrade avro version to 1.12.0 (#24617)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 36cd2c2ab32c32d6b52f6ce286fe92baaff3fd0f
Author: Hang Chen <ch...@apache.org>
AuthorDate: Wed Aug 13 22:38:54 2025 -0700
[improve][broker] Upgrade avro version to 1.12.0 (#24617)
Co-authored-by: Jiwe Guo <te...@apache.org>
---
distribution/server/src/assemble/LICENSE.bin.txt | 4 ++--
distribution/shell/src/assemble/LICENSE.bin.txt | 4 ++--
pom.xml | 2 +-
.../service/schema/JsonSchemaCompatibilityCheck.java | 3 +--
.../schema/validator/SchemaDataValidatorTest.java | 16 +++++++++++++++-
.../pulsar/client/impl/schema/util/SchemaUtil.java | 3 ++-
.../pulsar/client/impl/schema/AvroSchemaTest.java | 17 -----------------
.../apache/pulsar/io/elasticsearch/JsonConverter.java | 4 ++--
.../apache/pulsar/io/kinesis/json/JsonConverter.java | 4 ++--
9 files changed, 27 insertions(+), 30 deletions(-)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
index a11acdf5abf..0857b67ef96 100644
--- a/distribution/server/src/assemble/LICENSE.bin.txt
+++ b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -465,8 +465,8 @@ The Apache Software License, Version 2.0
* zt-zip
- org.zeroturnaround-zt-zip-1.17.jar
* Apache Avro
- - org.apache.avro-avro-1.11.4.jar
- - org.apache.avro-avro-protobuf-1.11.4.jar
+ - org.apache.avro-avro-1.12.0.jar
+ - org.apache.avro-avro-protobuf-1.12.0.jar
* Apache Curator
- org.apache.curator-curator-client-5.7.1.jar
- org.apache.curator-curator-framework-5.7.1.jar
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
index 28be70a6dc9..f2dfe800013 100644
--- a/distribution/shell/src/assemble/LICENSE.bin.txt
+++ b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -417,8 +417,8 @@ The Apache Software License, Version 2.0
* Google Error Prone Annotations - error_prone_annotations-2.38.0.jar
* Javassist -- javassist-3.25.0-GA.jar
* Apache Avro
- - avro-1.11.4.jar
- - avro-protobuf-1.11.4.jar
+ - avro-1.12.0.jar
+ - avro-protobuf-1.12.0.jar
* RE2j -- re2j-1.8.jar
* Spotify completable-futures -- completable-futures-0.3.6.jar
* RoaringBitmap -- RoaringBitmap-1.2.0.jar
diff --git a/pom.xml b/pom.xml
index 52066a87d4c..f76ebecdb1e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -225,7 +225,7 @@ flexible messaging model and an intuitive client API.</description>
<kafka-client.version>3.9.0</kafka-client.version>
<rabbitmq-client.version>5.18.0</rabbitmq-client.version>
<aws-sdk.version>1.12.638</aws-sdk.version>
- <avro.version>1.11.4</avro.version>
+ <avro.version>1.12.0</avro.version>
<joda.version>2.10.10</joda.version>
<jclouds.version>2.6.0</jclouds.version>
<guice.version>5.1.0</guice.version>
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java
index 476e5f272e4..cbdca3c4732 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java
@@ -23,7 +23,6 @@ import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
import java.io.IOException;
import org.apache.avro.Schema;
-import org.apache.avro.SchemaParseException;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.protocol.schema.SchemaData;
@@ -96,7 +95,7 @@ public class JsonSchemaCompatibilityCheck extends AvroSchemaBasedCompatibilityCh
fromParser.setValidateDefaults(false);
Schema fromSchema = fromParser.parse(new String(schemaData.getData(), UTF_8));
return true;
- } catch (SchemaParseException e) {
+ } catch (Exception e) {
return false;
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java
index 021fd4a9b40..302e5879d28 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java
@@ -20,6 +20,8 @@ package org.apache.pulsar.broker.service.schema.validator;
import static java.nio.charset.StandardCharsets.UTF_8;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
import com.fasterxml.jackson.module.jsonSchema.JsonSchemaGenerator;
import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException;
import org.apache.pulsar.client.api.Schema;
@@ -131,7 +133,19 @@ public class SchemaDataValidatorTest {
new JsonSchemaGenerator(mapper)
.generateSchema(Foo.class)))
.build();
- SchemaDataValidator.validateSchemaData(data);
+ ObjectReader reader = ObjectMapperFactory.getMapper().reader().forType(JsonSchema.class);
+ try {
+ org.apache.avro.Schema.Parser avroSchemaParser = new org.apache.avro.Schema.Parser();
+ avroSchemaParser.setValidateDefaults(false);
+ org.apache.avro.Schema schema = avroSchemaParser.parse(new String(data.getData(), UTF_8));
+ } catch (Exception e) {
+ try {
+ reader.readValue(data.getData());
+ } catch (Exception ioe) {
+ throw new InvalidSchemaDataException("Invalid schema definition data for "
+ + data.getType() + " schema", ioe);
+ }
+ }
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java
index 9c8d7fd511c..7af45734b89 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl.schema.util;
import static java.nio.charset.StandardCharsets.UTF_8;
import java.lang.reflect.Field;
+import org.apache.avro.NameValidator;
import org.apache.avro.Schema;
import org.apache.avro.reflect.ReflectData;
import org.apache.commons.lang3.StringUtils;
@@ -41,7 +42,7 @@ public class SchemaUtil {
}
public static Schema parseAvroSchema(String schemaJson) {
- final Schema.Parser parser = new Schema.Parser();
+ final Schema.Parser parser = new Schema.Parser(NameValidator.NO_VALIDATION);
parser.setValidateDefaults(false);
return parser.parse(schemaJson);
}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java
index 3dedf3c978c..e0a6cdc54fc 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java
@@ -506,15 +506,6 @@ public class AvroSchemaTest {
Assert.assertEquals(decodeWithJson, decodeWithPojo);
Assert.assertEquals(Instant.class, decodeWithJson.getValue().getClass());
-
- AvroSchema<TimestampPojo> schemaWithJsonDefNoClassLoader =
- AvroSchema.of(SchemaDefinition.<TimestampPojo>builder()
- .withJsonDef(schemaDefinition)
- .withJSR310ConversionEnabled(false).build());
-
- TimestampPojo decodeWithJsonNoClassLoader = schemaWithJsonDefNoClassLoader.decode(encode);
- Assert.assertNotEquals(decodeWithJsonNoClassLoader, decodeWithPojo);
- Assert.assertNotEquals(Instant.class, decodeWithJsonNoClassLoader.getValue().getClass());
}
@Test
@@ -540,14 +531,6 @@ public class AvroSchemaTest {
Assert.assertEquals(decodeWithJson, decodeWithPojo);
Assert.assertEquals(Instant.class, decodeWithJson.getValue().getClass());
- AvroSchema<TimestampPojo> schemaWithJsonDefNoClassLoader =
- AvroSchema.of(SchemaDefinition.<TimestampPojo>builder()
- .withJsonDef(schemaDefinition)
- .withJSR310ConversionEnabled(true).build());
-
- TimestampPojo decodeWithJsonNoClassLoader = schemaWithJsonDefNoClassLoader.decode(encode);
- Assert.assertNotEquals(decodeWithJsonNoClassLoader, decodeWithPojo);
- Assert.assertNotEquals(Instant.class, decodeWithJsonNoClassLoader.getValue().getClass());
}
@Data
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
index 01958cac58e..c3710485f9c 100644
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/JsonConverter.java
@@ -95,8 +95,8 @@ public class JsonConverter {
Schema elementSchema = schema.getElementType();
ArrayNode arrayNode = jsonNodeFactory.arrayNode();
Object[] iterable;
- if (value instanceof GenericData.Array) {
- iterable = ((GenericData.Array) value).toArray();
+ if (value instanceof GenericData.AbstractArray) {
+ iterable = ((GenericData.AbstractArray) value).toArray();
} else {
iterable = (Object[]) value;
}
diff --git a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/json/JsonConverter.java b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/json/JsonConverter.java
index a44d081cd2c..7b827b43542 100644
--- a/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/json/JsonConverter.java
+++ b/pulsar-io/kinesis/src/main/java/org/apache/pulsar/io/kinesis/json/JsonConverter.java
@@ -99,8 +99,8 @@ public class JsonConverter {
Schema elementSchema = schema.getElementType();
ArrayNode arrayNode = jsonNodeFactory.arrayNode();
Object[] iterable;
- if (value instanceof GenericData.Array) {
- iterable = ((GenericData.Array) value).toArray();
+ if (value instanceof GenericData.AbstractArray) {
+ iterable = ((GenericData.AbstractArray) value).toArray();
} else {
iterable = (Object[]) value;
}
(pulsar) 06/06: [improve][io] Add dependency file name information to error message when .nar file validation fails with ZipException (#24604)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 796603ea1b00e36d5833d0eef1e2f3bce301b7d3
Author: Phineas <62...@users.noreply.github.com>
AuthorDate: Fri Aug 8 00:32:54 2025 +0800
[improve][io] Add dependency file name information to error message when .nar file validation fails with ZipException (#24604)
Co-authored-by: Lari Hotari <lh...@apache.org>
---
.../PulsarFunctionTestTemporaryDirectory.java | 4 ++++
.../apache/pulsar/io/AbstractPulsarE2ETest.java | 3 ++-
.../apache/pulsar/io/PulsarFunctionE2ETest.java | 23 ++++++++++++++++++++++
.../functions/utils/FunctionFilePackage.java | 3 ++-
4 files changed, 31 insertions(+), 2 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTestTemporaryDirectory.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTestTemporaryDirectory.java
index 336aca37439..cd8a64e662b 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTestTemporaryDirectory.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTestTemporaryDirectory.java
@@ -80,4 +80,8 @@ public class PulsarFunctionTestTemporaryDirectory {
Assert.assertEquals(foundFiles.length, 0, "Temporary files left over: "
+ Arrays.asList(foundFiles));
}
+
+ public File getTempDirectory() {
+ return tempDirectory;
+ }
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java
index 74f76e7eb59..5b1a66fafb4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/AbstractPulsarE2ETest.java
@@ -313,7 +313,8 @@ public abstract class AbstractPulsarE2ETest {
workerConfig.setAuthorizationEnabled(true);
List<String> urlPatterns =
- List.of(getPulsarApiExamplesJar().getParentFile().toURI() + ".*", "http://127\\.0\\.0\\.1:.*");
+ List.of(getPulsarApiExamplesJar().getParentFile().toURI() + ".*", "http://127\\.0\\.0\\.1:.*",
+ tempDirectory.getTempDirectory().toURI() + ".*");
workerConfig.setAdditionalEnabledConnectorUrlPatterns(urlPatterns);
workerConfig.setAdditionalEnabledFunctionsUrlPatterns(urlPatterns);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
index e3114b2dd4b..bb3c969b5d4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.io;
import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically;
import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarApiExamplesJar;
import static org.apache.pulsar.functions.worker.PulsarFunctionLocalRunTest.getPulsarApiExamplesNar;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
@@ -32,6 +33,7 @@ import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.File;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
@@ -68,6 +70,7 @@ import org.apache.pulsar.functions.worker.TestPulsarFunctionUtils;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.Test;
+import org.zeroturnaround.zip.ZipUtil;
/**
* Test Pulsar sink on function.
@@ -203,6 +206,26 @@ public class PulsarFunctionE2ETest extends AbstractPulsarE2ETest {
testE2EPulsarFunction(jarFilePathUrl);
}
+ @Test(timeOut = 20000)
+ public void testE2EPulsarFunctionWithInvalidJarDependencyInNarFile() throws Exception {
+ File narFile = getPulsarApiExamplesNar();
+ File invalidNarFile = File.createTempFile("invalidJarDependency", ".nar", tempDirectory.getTempDirectory());
+ try {
+ // Add an invalid dependency to the nar file
+ ZipUtil.addEntry(narFile, "META-INF/bundled-dependencies/invalid-dependency.jar",
+ "Invalid jar content".getBytes(), invalidNarFile);
+ String jarFilePathUrl = invalidNarFile.toURI().toString();
+ testE2EPulsarFunction(jarFilePathUrl);
+ fail("Expected PulsarAdminException to be thrown due to invalid jar dependency in nar file");
+ } catch (PulsarAdminException e) {
+ assertThat(e.getMessage())
+ .contains("META-INF/bundled-dependencies/invalid-dependency.jar' failed due to zip END header not "
+ + "found");
+ } finally {
+ invalidNarFile.delete();
+ }
+ }
+
@Test(timeOut = 40000)
public void testE2EPulsarFunctionWithUrl() throws Exception {
testE2EPulsarFunction(fileServer.getUrl("/pulsar-functions-api-examples.jar"));
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionFilePackage.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionFilePackage.java
index 8224de32521..44e33410bd6 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionFilePackage.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionFilePackage.java
@@ -79,7 +79,8 @@ public class FunctionFilePackage implements AutoCloseable, ValidatableFunctionPa
}
classFileLocators.add(locator);
} catch (IOException e) {
- throw new UncheckedIOException(e);
+ throw new UncheckedIOException("Loading '" + classpath + "' failed due to " + e.getMessage(),
+ e);
}
}
}