database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Avoid 'No more data to read' error when handling stream load RPC (#354)

Browse Source 
* 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
This commit is contained in:
Mingyu Chen
2018-11-27 14:18:41 +08:00
committed by ZHAO Chun
parent dedfccfaf5
commit cddd864d83
2 changed files with 36 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
fe/src/main/java/org/apache/doris/common/UserException.java
View File

@ -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
fe/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
View File

@ -396,11 +396,11 @@ public class FrontendServiceImpl implements FrontendService.Iface {
} 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();
}
@ -465,7 +465,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
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;
}
@ -474,7 +474,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
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;
}
@ -554,12 +554,12 @@ public class FrontendServiceImpl implements FrontendService.Iface {
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;
}
@ -576,11 +576,17 @@ public class FrontendServiceImpl implements FrontendService.Iface {
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;
}
@ -624,12 +630,16 @@ public class FrontendServiceImpl implements FrontendService.Iface {
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;
}
@ -674,6 +684,11 @@ public class FrontendServiceImpl implements FrontendService.Iface {
} 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;
@ -704,6 +719,11 @@ public class FrontendServiceImpl implements FrontendService.Iface {
} 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;
}