From cddd864d8307ced7099ab00a66039256143fc9e0 Mon Sep 17 00:00:00 2001 From: Mingyu Chen Date: Tue, 27 Nov 2018 14:18:41 +0800 Subject: [PATCH] Avoid 'No more data to read' error when handling stream load RPC (#354) * 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 --- .../apache/doris/common/UserException.java | 10 +++-- .../doris/service/FrontendServiceImpl.java | 40 ++++++++++++++----- 2 files changed, 36 insertions(+), 14 deletions(-) diff --git a/fe/src/main/java/org/apache/doris/common/UserException.java b/fe/src/main/java/org/apache/doris/common/UserException.java index 7f04a08f92..7add338f93 100644 --- a/fe/src/main/java/org/apache/doris/common/UserException.java +++ b/fe/src/main/java/org/apache/doris/common/UserException.java @@ -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)); } } diff --git a/fe/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/src/main/java/org/apache/doris/service/FrontendServiceImpl.java index 9c6d2668c0..72ca0f9c4e 100644 --- a/fe/src/main/java/org/apache/doris/service/FrontendServiceImpl.java +++ b/fe/src/main/java/org/apache/doris/service/FrontendServiceImpl.java @@ -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; }