You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/09/12 20:43:09 UTC
(pulsar) branch master updated: [fix][fn] Bound function package downloads from BookKeeper and stop SchedulerManager.close() from hanging on a stuck scheduling round (#26556)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0ad91f147bbb [fix][fn] Bound function package downloads from BookKeeper and stop SchedulerManager.close() from hanging on a stuck scheduling round (#26556)
0ad91f147bbb is described below
commit 0ad91f147bbb1911feaf5b0af9af449882261a39
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Sat Sep 12 23:43:02 2026 +0300
[fix][fn] Bound function package downloads from BookKeeper and stop SchedulerManager.close() from hanging on a stuck scheduling round (#26556)
---
.../worker/WorkerUtilsBookKeeperPackageTest.java | 144 +++++++++++++++++++
.../pulsar/functions/worker/SchedulerManager.java | 81 ++++++++---
.../pulsar/functions/worker/WorkerUtils.java | 4 +-
.../functions/worker/dlog/DLInputStream.java | 152 +++++++++++++--------
.../functions/worker/SchedulerManagerTest.java | 48 +++++++
.../functions/worker/dlog/DLInputStreamTest.java | 149 +++++++++++++++++---
6 files changed, 484 insertions(+), 94 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/WorkerUtilsBookKeeperPackageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/WorkerUtilsBookKeeperPackageTest.java
new file mode 100644
index 000000000000..0b1852760e37
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/WorkerUtilsBookKeeperPackageTest.java
@@ -0,0 +1,144 @@
+/*
+ * 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.functions.worker;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.time.Duration;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.distributedlog.api.DistributedLogManager;
+import org.apache.distributedlog.api.namespace.Namespace;
+import org.apache.distributedlog.api.namespace.NamespaceBuilder;
+import org.apache.distributedlog.exceptions.LogNotFoundException;
+import org.apache.pulsar.common.conf.InternalConfigurationData;
+import org.apache.pulsar.functions.worker.dlog.DLInputStream;
+import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Tests uploading and downloading function packages to and from BookKeeper with the DistributedLog namespace that
+ * the function worker uses.
+ */
+@Test(groups = "functions-worker")
+public class WorkerUtilsBookKeeperPackageTest {
+ private static final Duration READ_TIMEOUT = Duration.ofSeconds(3);
+
+ private LocalBookkeeperEnsemble bkEnsemble;
+ private Namespace dlogNamespace;
+
+ @BeforeClass
+ void setup() throws Exception {
+ bkEnsemble = new LocalBookkeeperEnsemble(1, 0);
+ bkEnsemble.start();
+ InternalConfigurationData internalConf = new InternalConfigurationData(
+ "127.0.0.1:" + bkEnsemble.getZookeeperPort(), null, "/ledgers", null, null);
+ URI dlogUri = WorkerUtils.initializeDlogNamespace(internalConf);
+ WorkerConfig workerConfig = new WorkerConfig();
+ workerConfig.setNumFunctionPackageReplicas(1);
+ dlogNamespace = NamespaceBuilder.newBuilder()
+ .conf(WorkerUtils.getDlogConf(workerConfig))
+ .clientId("function-worker-test")
+ .uri(dlogUri)
+ .build();
+ }
+
+ @AfterClass(alwaysRun = true)
+ void cleanup() throws Exception {
+ if (dlogNamespace != null) {
+ dlogNamespace.close();
+ dlogNamespace = null;
+ }
+ if (bkEnsemble != null) {
+ bkEnsemble.stop();
+ bkEnsemble = null;
+ }
+ }
+
+ private byte[] download(String packagePath) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ try (DistributedLogManager dlm = dlogNamespace.openLog(packagePath);
+ InputStream in = new DLInputStream(dlm, READ_TIMEOUT)) {
+ in.transferTo(out);
+ }
+ return out.toByteArray();
+ }
+
+ @Test
+ public void testUploadAndDownload() throws Exception {
+ byte[] data = new byte[3_000_000];
+ ThreadLocalRandom.current().nextBytes(data);
+ WorkerUtils.uploadToBookKeeper(dlogNamespace, new ByteArrayInputStream(data), "tenant/ns/fn/complete");
+
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ WorkerUtils.downloadFromBookkeeper(dlogNamespace, out, "tenant/ns/fn/complete");
+ assertThat(out.toByteArray()).isEqualTo(data);
+ }
+
+ @Test
+ public void testDownloadMissingPackageFails() {
+ assertThatThrownBy(() -> download("tenant/ns/fn/missing"))
+ .isInstanceOf(LogNotFoundException.class);
+ }
+
+ @Test
+ public void testDownloadDeletedPackageFails() throws Exception {
+ WorkerUtils.uploadToBookKeeper(dlogNamespace, new ByteArrayInputStream(new byte[1000]),
+ "tenant/ns/fn/deleted");
+ WorkerUtils.deleteFromBookkeeper(dlogNamespace, "tenant/ns/fn/deleted");
+ assertThatThrownBy(() -> download("tenant/ns/fn/deleted"))
+ .isInstanceOf(LogNotFoundException.class);
+ }
+
+ /**
+ * A log stream that exists but has never been written to has no log segments to read. The download must fail
+ * after the read timeout instead of waiting forever for records to be written.
+ */
+ @Test
+ public void testDownloadEmptyPackageTimesOut() throws Exception {
+ dlogNamespace.createLog("tenant/ns/fn/empty");
+ assertThatThrownBy(() -> download("tenant/ns/fn/empty"))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Timed out");
+ }
+
+ /**
+ * A log stream of an upload that failed before {@link org.apache.pulsar.functions.worker.dlog.DLOutputStream}
+ * wrote the end-of-stream marker. The download must fail after the read timeout instead of waiting forever
+ * for the marker.
+ */
+ @Test
+ public void testDownloadIncompletePackageTimesOut() throws Exception {
+ byte[] data = new byte[1000];
+ try (DistributedLogManager dlm = dlogNamespace.openLog("tenant/ns/fn/incomplete");
+ var writer = dlm.getAppendOnlyStreamWriter()) {
+ writer.write(data);
+ writer.force(false);
+ }
+ assertThatThrownBy(() -> download("tenant/ns/fn/incomplete"))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Timed out");
+ }
+}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
index 6ddae690668e..e715c5098814 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java
@@ -116,6 +116,10 @@ public class SchedulerManager implements AutoCloseable {
AtomicBoolean isCompactionNeeded = new AtomicBoolean(false);
private static final long DEFAULT_ADMIN_API_BACKOFF_SEC = 60;
+ // how long close() waits for an in-progress scheduling round to complete before interrupting it, and then for
+ // the interrupted round to unwind
+ private static final long DEFAULT_CLOSE_SCHEDULING_ROUND_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(10);
+ private volatile long closeSchedulingRoundTimeoutMs = DEFAULT_CLOSE_SCHEDULING_ROUND_TIMEOUT_MS;
public static final String HEARTBEAT_TENANT = "pulsar-function";
public static final String HEARTBEAT_NAMESPACE = "heartbeat";
@@ -209,9 +213,16 @@ public class SchedulerManager implements AutoCloseable {
try {
runnable.run();
} catch (Throwable th) {
- log.error().attr("context", errMsg)
- .log("Encountered error when invoking scheduler");
- errorNotifier.triggerError(th);
+ if (!isRunning) {
+ // close() interrupts a scheduling round that doesn't complete in time; the
+ // failure of a round that was cut short by closing is not an error of the worker
+ log.warn().attr("context", errMsg).exception(th)
+ .log("Scheduler manager was closed while invoking the scheduler");
+ } else {
+ log.error().attr("context", errMsg).exception(th)
+ .log("Encountered error when invoking scheduler");
+ errorNotifier.triggerError(th);
+ }
}
}
} finally {
@@ -768,33 +779,71 @@ public class SchedulerManager implements AutoCloseable {
}
@Override
- public synchronized void close() {
+ public void close() {
log.info("Closing scheduler manager");
- // make sure we are not closing while a scheduling is being calculated
- schedulerLock.lock();
- try {
+ final ThreadPoolExecutor executor;
+ final ScheduledExecutorService scheduledExecutor;
+ final Producer<byte[]> producer;
+ // don't hold the monitor while waiting for the scheduler lock: a scheduling round holds the scheduler lock
+ // while calling synchronized methods of this class
+ synchronized (this) {
isRunning = false;
+ executor = executorService;
+ scheduledExecutor = scheduledExecutorService;
+ producer = exclusiveProducer;
+ }
- if (scheduledExecutorService != null) {
- scheduledExecutorService.shutdown();
- }
+ if (scheduledExecutor != null) {
+ scheduledExecutor.shutdown();
+ }
+ if (executor != null) {
+ // stop accepting new scheduling rounds, a round that is already running keeps running
+ executor.shutdown();
+ }
- if (executorService != null) {
- executorService.shutdown();
+ // make sure we are not closing the assignment producer while a scheduling round is being calculated.
+ // A round can block for a long time (for example while downloading a function package), so don't wait for
+ // it indefinitely: interrupt it and continue closing.
+ boolean locked = tryLockSchedulerLock();
+ if (!locked && executor != null) {
+ log.warn().attr("timeoutMs", closeSchedulingRoundTimeoutMs)
+ .log("Scheduling round did not complete in time, interrupting it");
+ executor.shutdownNow();
+ locked = tryLockSchedulerLock();
+ if (!locked) {
+ log.warn().attr("timeoutMs", closeSchedulingRoundTimeoutMs)
+ .log("Interrupted scheduling round did not complete in time, closing anyway");
}
-
- if (exclusiveProducer != null) {
+ }
+ try {
+ if (producer != null) {
try {
- exclusiveProducer.close();
+ producer.close();
} catch (PulsarClientException e) {
log.warn().exception(e).log("Failed to shutdown scheduler manager assignment producer");
}
}
} finally {
- schedulerLock.unlock();
+ if (locked) {
+ schedulerLock.unlock();
+ }
+ }
+ }
+
+ private boolean tryLockSchedulerLock() {
+ try {
+ return schedulerLock.tryLock(closeSchedulingRoundTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
}
}
+ @VisibleForTesting
+ void setCloseSchedulingRoundTimeoutMs(long closeSchedulingRoundTimeoutMs) {
+ this.closeSchedulingRoundTimeoutMs = closeSchedulingRoundTimeoutMs;
+ }
+
static String checkHeartBeatFunction(Instance funInstance) {
if (funInstance.getFunctionMetaData() != null
&& funInstance.getFunctionMetaData().getFunctionDetails() != null) {
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerUtils.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerUtils.java
index 477c134237cb..8fc49bbe9feb 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerUtils.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerUtils.java
@@ -126,8 +126,8 @@ public final class WorkerUtils {
OutputStream outputStream,
String packagePath) throws IOException {
log.info().attr("packagePath", packagePath).log("Downloading from BK");
- DistributedLogManager dlm = namespace.openLog(packagePath);
- try (InputStream in = new DLInputStream(dlm)) {
+ try (DistributedLogManager dlm = namespace.openLog(packagePath);
+ InputStream in = new DLInputStream(dlm)) {
int read = 0;
byte[] bytes = new byte[1024];
while ((read = in.read(bytes)) != -1) {
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/dlog/DLInputStream.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/dlog/DLInputStream.java
index f009f57d1aed..2f9d063e9ced 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/dlog/DLInputStream.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/dlog/DLInputStream.java
@@ -20,35 +20,47 @@ package org.apache.pulsar.functions.worker.dlog;
import java.io.IOException;
import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.distributedlog.DLSN;
import org.apache.distributedlog.LogRecordWithDLSN;
+import org.apache.distributedlog.api.AsyncLogReader;
import org.apache.distributedlog.api.DistributedLogManager;
-import org.apache.distributedlog.api.LogReader;
import org.apache.distributedlog.exceptions.EndOfStreamException;
/**
* DistributedLog Input Stream.
+ *
+ * <p>Reads the records of a log stream until the end-of-stream marker written by {@link DLOutputStream#close()}.
+ * A log stream is a tailing log: a reader of a stream that has no end-of-stream marker (for example because the
+ * writer failed before completing the upload, or the stream was never written to) waits for more records to arrive
+ * forever. To avoid blocking the calling thread indefinitely, waiting for the next records is bounded by a read
+ * timeout and fails with an {@link IOException} when it expires.
*/
public class DLInputStream extends InputStream {
- private LogRecordWithInputStream currentLogRecord = null;
- private final DistributedLogManager dlm;
- private LogReader reader;
- private boolean eos = false;
-
- // Cache the input stream for a log record.
- private static class LogRecordWithInputStream {
- private final InputStream payloadStream;
-
- LogRecordWithInputStream(LogRecordWithDLSN logRecord) {
- this.payloadStream = logRecord.getPayLoadInputStream();
- }
+ /**
+ * Default maximum time to wait for the next records of the log stream to become available.
+ */
+ public static final Duration DEFAULT_READ_TIMEOUT = Duration.ofSeconds(60);
- InputStream getPayLoadInputStream() {
- return payloadStream;
- }
+ private static final int READ_BATCH_SIZE = 100;
- }
+ private final DistributedLogManager dlm;
+ private final AsyncLogReader reader;
+ private final long readTimeoutMs;
+ private final Deque<LogRecordWithDLSN> pendingRecords = new ArrayDeque<>();
+ // payload of the log record that is currently being consumed
+ private InputStream currentPayload;
+ private boolean eos = false;
/**
* Construct DistributedLog input stream.
@@ -56,38 +68,73 @@ public class DLInputStream extends InputStream {
* @param dlm the Distributed Log Manager to access the stream
*/
public DLInputStream(DistributedLogManager dlm) throws IOException {
- this.dlm = dlm;
- reader = dlm.getInputStream(DLSN.InitialDLSN);
+ this(dlm, DEFAULT_READ_TIMEOUT);
}
/**
- * Get input stream representing next entry in the
- * ledger.
+ * Construct DistributedLog input stream.
*
- * @return input stream, or null if no more entries
+ * @param dlm the Distributed Log Manager to access the stream
+ * @param readTimeout maximum time to wait for the next records of the stream to become available
*/
- private LogRecordWithInputStream nextLogRecord() throws IOException {
+ public DLInputStream(DistributedLogManager dlm, Duration readTimeout) throws IOException {
+ this.dlm = dlm;
+ this.readTimeoutMs = readTimeout.toMillis();
+ this.reader = await(dlm.openAsyncLogReader(DLSN.InitialDLSN), "open a reader for");
+ }
+
+ private <T> T await(CompletableFuture<T> future, String operation) throws IOException {
try {
- return nextLogRecord(reader);
- } catch (EndOfStreamException e) {
- eos = true;
- return null;
+ return future.get(readTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(false);
+ throw new IOException("Timed out after " + readTimeoutMs + " ms waiting to " + operation + " log stream "
+ + dlm.getStreamName(), e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException("Interrupted while waiting to " + operation + " log stream "
+ + dlm.getStreamName());
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
+ throw new IOException("Failed to " + operation + " log stream " + dlm.getStreamName(), cause);
}
}
- private static LogRecordWithInputStream nextLogRecord(LogReader reader) throws IOException {
- LogRecordWithDLSN record = reader.readNext(false);
+ /**
+ * Get the next log record of the stream.
+ *
+ * @return the next log record, or null when the end of the stream has been reached
+ */
+ private LogRecordWithDLSN nextLogRecord() throws IOException {
+ if (pendingRecords.isEmpty() && !eos) {
+ try {
+ List<LogRecordWithDLSN> records = await(reader.readBulk(READ_BATCH_SIZE), "read");
+ pendingRecords.addAll(records);
+ } catch (EndOfStreamException e) {
+ eos = true;
+ }
+ }
+ return pendingRecords.pollFirst();
+ }
- if (null != record) {
- return new LogRecordWithInputStream(record);
- } else {
- record = reader.readNext(false);
- if (null != record) {
- return new LogRecordWithInputStream(record);
- } else {
+ /**
+ * Get the payload of the log record that is currently being consumed, moving on to the next log record when the
+ * current payload has been fully consumed.
+ *
+ * @return the payload input stream, or null when the end of the stream has been reached
+ */
+ private InputStream currentPayload() throws IOException {
+ if (currentPayload == null) {
+ LogRecordWithDLSN record = nextLogRecord();
+ if (record == null) {
return null;
}
+ currentPayload = record.getPayLoadInputStream();
}
+ return currentPayload;
}
@Override
@@ -102,35 +149,32 @@ public class DLInputStream extends InputStream {
@Override
public int read(byte[] b, int off, int len) throws IOException {
- if (eos) {
- return -1;
+ Objects.checkFromIndexSize(off, len, b.length);
+ if (len == 0) {
+ return 0;
}
-
int read = 0;
- if (currentLogRecord == null) {
- currentLogRecord = nextLogRecord();
- if (currentLogRecord == null) {
- return read;
- }
- }
-
while (read < len) {
- int thisread = currentLogRecord.getPayLoadInputStream().read(b, off + read, len - read);
+ InputStream payload = currentPayload();
+ if (payload == null) {
+ break;
+ }
+ int thisread = payload.read(b, off + read, len - read);
if (thisread == -1) {
- currentLogRecord = nextLogRecord();
- if (currentLogRecord == null) {
- return read;
- }
+ currentPayload = null;
} else {
read += thisread;
}
}
- return read;
+ return read == 0 ? -1 : read;
}
@Override
public void close() throws IOException {
- reader.close();
- dlm.close();
+ try {
+ await(reader.asyncClose(), "close the reader of");
+ } finally {
+ dlm.close();
+ }
}
}
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
index 2661cc288d1f..42cc0788d136 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/SchedulerManagerTest.java
@@ -24,6 +24,7 @@ import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyBoolean;
import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
@@ -41,12 +42,14 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
import lombok.CustomLog;
import lombok.val;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
@@ -951,6 +954,51 @@ public class SchedulerManagerTest {
}
}
+ @Test(timeOut = 60000)
+ public void testCloseInterruptsStuckSchedulingRound() throws Exception {
+ List<FunctionMetaData> functionMetaDataList = new LinkedList<>();
+ functionMetaDataList.add(createFunctionMetaData("tenant-1", "namespace-1", "func-1", 1, 1));
+ doReturn(functionMetaDataList).when(functionMetaDataManager).getAllFunctionMetaData();
+ ThreadRuntimeFactory factory = mock(ThreadRuntimeFactory.class);
+ doReturn(factory).when(functionRuntimeManager).getRuntimeFactory();
+ doReturn(new HashMap<>()).when(functionRuntimeManager).getCurrentAssignments();
+ List<WorkerInfo> workerInfoList = new LinkedList<>();
+ workerInfoList.add(WorkerInfo.of("worker-1", "workerHostname-1", 5000));
+ doReturn(workerInfoList).when(membershipManager).getCurrentMembership();
+ doReturn(true).when(leaderService).isLeader();
+
+ // processing the new assignment blocks until interrupted, like a function package download that never
+ // completes
+ CountDownLatch roundStuck = new CountDownLatch(1);
+ AtomicBoolean roundInterrupted = new AtomicBoolean();
+ doAnswer(invocation -> {
+ roundStuck.countDown();
+ try {
+ new CountDownLatch(1).await();
+ } catch (InterruptedException e) {
+ roundInterrupted.set(true);
+ throw new RuntimeException(e);
+ }
+ return null;
+ }).when(functionRuntimeManager).processAssignment(any(Assignment.class));
+
+ schedulerManager.setCloseSchedulingRoundTimeoutMs(500);
+ schedulerManager.initialize(schedulerManager.acquireExclusiveWrite(() -> true));
+ schedulerManager.schedule();
+ assertTrue(roundStuck.await(30, TimeUnit.SECONDS));
+
+ long closeStart = System.nanoTime();
+ schedulerManager.close();
+ long closeDurationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - closeStart);
+ assertTrue(closeDurationMs < 10000, "close() took " + closeDurationMs + " ms");
+ assertTrue(roundInterrupted.get());
+ verify(producer, times(1)).close();
+ // the failure of the interrupted scheduling round is not a worker error
+ verify(errorNotifier, times(0)).triggerError(any());
+ assertTrue(schedulerManager.getSchedulerLock().tryLock(30, TimeUnit.SECONDS));
+ schedulerManager.getSchedulerLock().unlock();
+ }
+
private void callSchedule() throws InterruptedException,
TimeoutException, ExecutionException, WorkerUtils.NotLeaderAnymore {
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/dlog/DLInputStreamTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/dlog/DLInputStreamTest.java
index 60b01391aeae..98058f773f31 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/dlog/DLInputStreamTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/dlog/DLInputStreamTest.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.functions.worker.dlog;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyBoolean;
+import static org.mockito.Mockito.anyInt;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -27,12 +29,19 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.AssertJUnit.assertEquals;
import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeoutException;
import org.apache.distributedlog.DLSN;
import org.apache.distributedlog.LogRecordWithDLSN;
+import org.apache.distributedlog.api.AsyncLogReader;
import org.apache.distributedlog.api.DistributedLogManager;
-import org.apache.distributedlog.api.LogReader;
import org.apache.distributedlog.exceptions.EndOfStreamException;
+import org.apache.distributedlog.exceptions.LogNotFoundException;
import org.testng.annotations.Test;
/**
@@ -40,22 +49,44 @@ import org.testng.annotations.Test;
*/
public class DLInputStreamTest {
+ private static DistributedLogManager mockDlm(AsyncLogReader reader) {
+ DistributedLogManager dlm = mock(DistributedLogManager.class);
+ when(dlm.getStreamName()).thenReturn("test-stream");
+ when(dlm.openAsyncLogReader(any(DLSN.class))).thenReturn(CompletableFuture.completedFuture(reader));
+ return dlm;
+ }
+
+ private static AsyncLogReader mockReader() {
+ AsyncLogReader reader = mock(AsyncLogReader.class);
+ when(reader.asyncClose()).thenReturn(CompletableFuture.completedFuture(null));
+ return reader;
+ }
+
+ private static LogRecordWithDLSN mockRecord(byte[] payload) {
+ LogRecordWithDLSN record = mock(LogRecordWithDLSN.class);
+ when(record.getPayLoadInputStream()).thenReturn(new ByteArrayInputStream(payload));
+ return record;
+ }
+
/**
* Test Case: reader hits eos (end of stream).
*/
@Test
public void testReadEos() throws Exception {
- DistributedLogManager dlm = mock(DistributedLogManager.class);
- LogReader reader = mock(LogReader.class);
- when(dlm.getInputStream(any(DLSN.class))).thenReturn(reader);
- when(reader.readNext(anyBoolean())).thenThrow(new EndOfStreamException("eos"));
+ AsyncLogReader reader = mockReader();
+ DistributedLogManager dlm = mockDlm(reader);
+ when(reader.readBulk(anyInt())).thenReturn(CompletableFuture.failedFuture(new EndOfStreamException("eos")));
byte[] b = new byte[1];
DLInputStream in = new DLInputStream(dlm);
- assertEquals("Should return 0 when reading an empty eos stream",
- 0, in.read(b, 0, 1));
assertEquals("Should return -1 when reading an empty eos stream",
-1, in.read(b, 0, 1));
+ assertEquals("Should keep returning -1 when reading an empty eos stream",
+ -1, in.read(b, 0, 1));
+ assertEquals("Should return -1 when reading a byte from an empty eos stream",
+ -1, in.read());
+ // the end of stream is reached only once
+ verify(reader, times(1)).readBulk(anyInt());
}
/**
@@ -63,15 +94,14 @@ public class DLInputStreamTest {
*/
@Test
public void testClose() throws Exception {
- DistributedLogManager dlm = mock(DistributedLogManager.class);
- LogReader reader = mock(LogReader.class);
- when(dlm.getInputStream(any(DLSN.class))).thenReturn(reader);
+ AsyncLogReader reader = mockReader();
+ DistributedLogManager dlm = mockDlm(reader);
DLInputStream in = new DLInputStream(dlm);
- verify(dlm, times(1)).getInputStream(eq(DLSN.InitialDLSN));
+ verify(dlm, times(1)).openAsyncLogReader(eq(DLSN.InitialDLSN));
in.close();
+ verify(reader, times(1)).asyncClose();
verify(dlm, times(1)).close();
- verify(reader, times(1)).close();
}
/**
@@ -79,17 +109,14 @@ public class DLInputStreamTest {
*/
@Test
public void testRead() throws Exception {
- DistributedLogManager dlm = mock(DistributedLogManager.class);
- LogReader reader = mock(LogReader.class);
- when(dlm.getInputStream(any(DLSN.class))).thenReturn(reader);
+ AsyncLogReader reader = mockReader();
+ DistributedLogManager dlm = mockDlm(reader);
byte[] data = "test-read".getBytes(StandardCharsets.UTF_8);
- LogRecordWithDLSN record = mock(LogRecordWithDLSN.class);
- when(record.getPayLoadInputStream())
- .thenReturn(new ByteArrayInputStream(data));
- when(reader.readNext(anyBoolean()))
- .thenReturn(record)
- .thenThrow(new EndOfStreamException("eos"));
+ LogRecordWithDLSN record = mockRecord(data);
+ when(reader.readBulk(anyInt()))
+ .thenReturn(CompletableFuture.completedFuture(List.of(record)))
+ .thenReturn(CompletableFuture.failedFuture(new EndOfStreamException("eos")));
DLInputStream in = new DLInputStream(dlm);
int numReads = 0;
@@ -101,4 +128,82 @@ public class DLInputStreamTest {
assertEquals(data.length, numReads);
}
+ /**
+ * Test Case: a read spans multiple records and batches of records.
+ */
+ @Test
+ public void testReadAcrossRecords() throws Exception {
+ AsyncLogReader reader = mockReader();
+ DistributedLogManager dlm = mockDlm(reader);
+
+ List<LogRecordWithDLSN> firstBatch = List.of(mockRecord("ab".getBytes(StandardCharsets.UTF_8)),
+ mockRecord("cd".getBytes(StandardCharsets.UTF_8)));
+ List<LogRecordWithDLSN> secondBatch = List.of(mockRecord("ef".getBytes(StandardCharsets.UTF_8)));
+ when(reader.readBulk(anyInt()))
+ .thenReturn(CompletableFuture.completedFuture(firstBatch))
+ .thenReturn(CompletableFuture.completedFuture(secondBatch))
+ .thenReturn(CompletableFuture.failedFuture(new EndOfStreamException("eos")));
+
+ try (InputStream in = new DLInputStream(dlm)) {
+ byte[] b = new byte[5];
+ assertThat(in.read(b, 0, 5)).isEqualTo(5);
+ assertThat(new String(b, StandardCharsets.UTF_8)).isEqualTo("abcde");
+ assertThat(in.read(b, 0, 5)).isEqualTo(1);
+ assertThat(b[0]).isEqualTo((byte) 'f');
+ assertThat(in.read(b, 0, 5)).isEqualTo(-1);
+ }
+ }
+
+ /**
+ * Test Case: the stream never delivers the next records (for example an incomplete upload without an
+ * end-of-stream marker). The read must fail after the read timeout instead of blocking forever.
+ */
+ @Test
+ public void testReadTimesOut() throws Exception {
+ AsyncLogReader reader = mockReader();
+ DistributedLogManager dlm = mockDlm(reader);
+ CompletableFuture<List<LogRecordWithDLSN>> neverCompleted = new CompletableFuture<>();
+ when(reader.readBulk(anyInt())).thenReturn(neverCompleted);
+
+ try (InputStream in = new DLInputStream(dlm, Duration.ofMillis(200))) {
+ assertThatThrownBy(() -> in.read(new byte[1], 0, 1))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Timed out")
+ .hasMessageContaining("test-stream")
+ .hasCauseInstanceOf(TimeoutException.class);
+ }
+ assertThat(neverCompleted).isCancelled();
+ }
+
+ /**
+ * Test Case: reading fails with the failure of the log stream read.
+ */
+ @Test
+ public void testReadFailure() throws Exception {
+ AsyncLogReader reader = mockReader();
+ DistributedLogManager dlm = mockDlm(reader);
+ when(reader.readBulk(anyInt()))
+ .thenReturn(CompletableFuture.failedFuture(new LogNotFoundException("deleted")));
+
+ try (InputStream in = new DLInputStream(dlm)) {
+ assertThatThrownBy(() -> in.read(new byte[1], 0, 1))
+ .isInstanceOf(LogNotFoundException.class)
+ .hasMessage("deleted");
+ }
+ }
+
+ /**
+ * Test Case: opening the reader fails.
+ */
+ @Test
+ public void testOpenReaderFailure() {
+ DistributedLogManager dlm = mock(DistributedLogManager.class);
+ when(dlm.getStreamName()).thenReturn("test-stream");
+ when(dlm.openAsyncLogReader(any(DLSN.class)))
+ .thenReturn(CompletableFuture.failedFuture(new LogNotFoundException("missing")));
+
+ assertThatThrownBy(() -> new DLInputStream(dlm))
+ .isInstanceOf(LogNotFoundException.class)
+ .hasMessage("missing");
+ }
}