Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Fix][Connector-V2] Fix doris sink can not be closed when stream load not read any data #6570

Merged
merged 1 commit into from
Mar 27, 2024
Merged

[Fix][Connector-V2] Fix doris sink can not be closed when stream load not read any data #6570

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
14 changes: 7 additions & 7 deletions ...oris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,9 +65,9 @@ public class DorisSinkWriter
private final int intervalTime;
private final DorisSerializer serializer;
private final CatalogTable catalogTable;
private final transient ScheduledExecutorService scheduledExecutorService;
private transient Thread executorThread;
private transient volatile Exception loadException = null;
private final ScheduledExecutorService scheduledExecutorService;
private Thread executorThread;
private volatile Exception loadException = null;

public DorisSinkWriter(
SinkWriter.Context context,
Expand Down Expand Up @@ -114,8 +114,6 @@ private void initializeLoad() {
} catch (Exception e) {
throw new DorisConnectorException(DorisConnectorErrorCode.STREAM_LOAD_FAILED, e);
}
// get main work thread.
executorThread = Thread.currentThread();
startLoad(labelGenerator.generateLabel(lastCheckpointId + 1));
// when uploading data in streaming mode, we need to regularly detect whether there are
// exceptions.
Expand All @@ -125,7 +123,7 @@ private void initializeLoad() {

@Override
public void write(SeaTunnelRow element) throws IOException {
checkLoadException();
checkLoadExceptionAndResetThread();
byte[] serialize =
serializer.serialize(
dorisConfig.isNeedsUnsupportedTypeCasting()
Expand Down Expand Up @@ -222,9 +220,11 @@ private void checkDone() {
}
}

private void checkLoadException() {
private void checkLoadExceptionAndResetThread() {
if (loadException != null) {
throw new RuntimeException("error while loading data.", loadException);
} else {
executorThread = Thread.currentThread();
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Before this PR, the executorThread value not right. So that stream load check error thread can not interupt write thread normally.

}
}

Expand Down