[Pulsar SQL] Fix PulsarRecordCursor deserialize issue.
@@ -44,7 +44,9 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
@@ -109,6 +111,7 @@ public class PulsarRecordCursor implements RecordCursor {
private final long splitSize;
private long entriesProcessed = 0;
private int partition = -1;
+ private volatile Throwable deserializingError;
private PulsarSqlSchemaInfoProvider schemaInfoProvider;
@@ -235,30 +238,36 @@ public class PulsarRecordCursor implements RecordCursor {
}
@VisibleForTesting
- class DeserializeEntries implements Runnable {
+ class DeserializeEntries extends Thread {
- protected boolean isRunning = false;
+ private final AtomicBoolean isRunning;
- private final Thread thread;
+ private final CompletableFuture<Void> closeHandle;
public DeserializeEntries() {
- this.thread = new Thread(this, "derserialize-thread-split-" + pulsarSplit.getSplitId());
+ super("deserialize-thread-split-" + pulsarSplit.getSplitId());
+ this.isRunning = new AtomicBoolean(false);
+ this.closeHandle = new CompletableFuture<>();
}
- public void interrupt() {
- isRunning = false;
- thread.interrupt();
+ @Override
+ public void start() {
+ if (isRunning.compareAndSet(false, true)) {
+ super.start();
+ }
}
- public void start() {
- this.thread.start();
+ public CompletableFuture<Void> close() {
+ if (isRunning.compareAndSet(true, false)) {
+ super.interrupt();
+ }
+ return closeHandle;
}
@Override
public void run() {
- isRunning = true;
- while (isRunning) {
-
+ try {
+ while (isRunning.get()) {
int read = entryQueue.drain(new MessagePassingQueue.Consumer<Entry>() {
@Override
public void accept(Entry entry) {
@@ -343,6 +352,12 @@ public class PulsarRecordCursor implements RecordCursor {
}
}
}
+ closeHandle.complete(null);
+ } catch (Throwable ex) {
+ log.error(ex, "Stop running DeserializeEntries");
+ closeHandle.completeExceptionally(ex);
+ throw ex;
+ }
}
}
@@ -467,6 +482,9 @@ public class PulsarRecordCursor implements RecordCursor {
if (readEntries == null) {
// start deserialize thread
deserializeEntries = new DeserializeEntries();
+ deserializeEntries.setUncaughtExceptionHandler((t, ex) -> {
+ deserializingError = ex;
+ });
deserializeEntries.start();
readEntries = new ReadEntries();
@@ -491,6 +509,8 @@ public class PulsarRecordCursor implements RecordCursor {
if (currentMessage != null) {
messageQueueCacheSizeAllocator.release(currentMessage.getData().readableBytes());
break;
+ } else if (deserializingError != null) {
+ throw new RuntimeException(deserializingError);
} else {
try {
Thread.sleep(1);
@@ -502,7 +522,7 @@ public class PulsarRecordCursor implements RecordCursor {
}
}
- //start time for deseralizing record
+ //start time for deserializing record
metricsTracker.start_RECORD_DESERIALIZE_TIME();
SchemaInfo schemaInfo = getSchemaInfo(pulsarSplit);
@@ -713,12 +733,12 @@ public class PulsarRecordCursor implements RecordCursor {
messageQueue.drain(RawMessage::release);
}
+ if (deserializeEntries != null) {
+ deserializeEntries.close().whenComplete((r, t) -> {
if (entryQueue != null) {
entryQueue.drain(Entry::release);
}
-
- if (deserializeEntries != null) {
- deserializeEntries.interrupt();
+ });
}
if (this.cursor != null) {
