Skip to content

Commit

Permalink
Avoid 'No more data to read' error when handling stream load RPC (#354)
Browse files Browse the repository at this point in the history 
* Avoid 'No more data to read' error when handling stream load rpc

1. Catch throwable of all stream load rpc.
2. Avoid setting null string as error msg of rpc result status.

* Change setError_msgs to addToError_msgs
  • Loading branch information
@morningman @imay
morningman authored and imay committed Nov 27, 2018
1 parent dedfccf commit cddd864
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 14 deletions.
10 changes: 6 additions & 4 deletions fe/src/main/java/org/apache/doris/common/UserException.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,24 +17,26 @@

package org.apache.doris.common;

import com.google.common.base.Strings;

/**
* Thrown for internal server errors.
*/
public class UserException extends Exception {
public UserException(String msg, Throwable cause) {
super(msg, cause);
super(Strings.nullToEmpty(msg), cause);
}

public UserException(Throwable cause) {
super(cause);
}

public UserException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
super(message, cause, enableSuppression, writableStackTrace);
public UserException(String msg, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
super(Strings.nullToEmpty(msg), cause, enableSuppression, writableStackTrace);
}

public UserException(String msg) {
super(msg);
super(Strings.nullToEmpty(msg));
}

}
40 changes: 30 additions & 10 deletions fe/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -396,11 +396,11 @@ public TFeResult miniLoad(TMiniLoadRequest request) throws TException {
} catch (UserException e) {
LOG.warn("add mini load error", e);
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.setError_msgs(Lists.newArrayList(e.getMessage()));
status.addToError_msgs(e.getMessage());
} catch (Throwable e) {
LOG.warn("unexpected exception when adding mini load", e);
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.setError_msgs(Lists.newArrayList(e.getMessage()));
status.addToError_msgs(Strings.nullToEmpty(e.getMessage()));
} finally {
ConnectContext.remove();
}
Expand Down Expand Up @@ -465,7 +465,7 @@ public TFeResult updateMiniEtlTaskStatus(TUpdateMiniEtlTaskStatusRequest request
String failMsg = "job does not exist. id: " + jobId;
LOG.warn(failMsg);
status.setStatus_code(TStatusCode.CANCELLED);
status.setError_msgs(Lists.newArrayList(failMsg));
status.addToError_msgs(failMsg);
return result;
}

Expand All @@ -474,7 +474,7 @@ public TFeResult updateMiniEtlTaskStatus(TUpdateMiniEtlTaskStatusRequest request
String failMsg = "task info does not exist. task id: " + taskId + ", job id: " + jobId;
LOG.warn(failMsg);
status.setStatus_code(TStatusCode.CANCELLED);
status.setError_msgs(Lists.newArrayList(failMsg));
status.addToError_msgs(failMsg);
return result;
}

Expand Down Expand Up @@ -554,12 +554,12 @@ public TFeResult loadCheck(TLoadCheckRequest request) throws TException {
request.getTbl(), request.getUser_ip(), PrivPredicate.LOAD);
} catch (UserException e) {
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.setError_msgs(Lists.newArrayList(e.getMessage()));
status.addToError_msgs(e.getMessage());
return result;
} catch (Throwable e) {
LOG.warn("catch unknown result.", e);
status.setStatus_code(TStatusCode.INTERNAL_ERROR);
status.setError_msgs(Lists.newArrayList(e.getMessage()));
status.addToError_msgs(Strings.nullToEmpty(e.getMessage()));
return result;
}

Expand All @@ -576,11 +576,17 @@ public TLoadTxnBeginResult loadTxnBegin(TLoadTxnBeginRequest request) throws TEx
result.setTxnId(loadTxnBeginImpl(request));
} catch (LabelAlreadyExistsException e) {
status.setStatus_code(TStatusCode.LABEL_ALREADY_EXISTS);
status.setError_msgs(Lists.newArrayList(e.getMessage()));
status.addToError_msgs(e.getMessage());
} catch (UserException e) {
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.setError_msgs(Lists.newArrayList(e.getMessage()));
status.addToError_msgs(e.getMessage());
} catch (Throwable e) {
LOG.warn("catch unknown result.", e);
status.setStatus_code(TStatusCode.INTERNAL_ERROR);
status.addToError_msgs(Strings.nullToEmpty(e.getMessage()));
return result;
}

return result;
}

Expand Down Expand Up @@ -624,12 +630,16 @@ public TLoadTxnCommitResult loadTxnCommit(TLoadTxnCommitRequest request) throws
if (!loadTxnCommitImpl(request)) {
// committed success but not visible
status.setStatus_code(TStatusCode.PUBLISH_TIMEOUT);
status.setError_msgs(
Lists.newArrayList("transaction commit successfully, BUT data will be visible later"));
status.addToError_msgs("transaction commit successfully, BUT data will be visible later");
}
} catch (UserException e) {
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.addToError_msgs(e.getMessage());
} catch (Throwable e) {
LOG.warn("catch unknown result.", e);
status.setStatus_code(TStatusCode.INTERNAL_ERROR);
status.addToError_msgs(Strings.nullToEmpty(e.getMessage()));
return result;
}
return result;
}
Expand Down Expand Up @@ -674,6 +684,11 @@ public TLoadTxnRollbackResult loadTxnRollback(TLoadTxnRollbackRequest request) t
} catch (UserException e) {
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.addToError_msgs(e.getMessage());
} catch (Throwable e) {
LOG.warn("catch unknown result.", e);
status.setStatus_code(TStatusCode.INTERNAL_ERROR);
status.addToError_msgs(Strings.nullToEmpty(e.getMessage()));
return result;
}

return result;
Expand Down Expand Up @@ -704,6 +719,11 @@ public TStreamLoadPutResult streamLoadPut(TStreamLoadPutRequest request) throws
} catch (UserException e) {
status.setStatus_code(TStatusCode.ANALYSIS_ERROR);
status.addToError_msgs(e.getMessage());
} catch (Throwable e) {
LOG.warn("catch unknown result.", e);
status.setStatus_code(TStatusCode.INTERNAL_ERROR);
status.addToError_msgs(Strings.nullToEmpty(e.getMessage()));
return result;
}
return result;
}
Expand Down

0 comments on commit cddd864

Please sign in to comment.