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

[improvement](spark-connector) Stream load http exception handling (#7514)

Browse Source 
Stream load http exception handling
This commit is contained in:
jiafeng.zhang
2022-01-09 16:54:55 +08:00
committed by GitHub
parent 3a8a85b739
commit 9aaa3f63f7
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
extension/spark-doris-connector/src/main/java/org/apache/doris/spark/DorisStreamLoad.java
View File

@ -157,10 +157,10 @@ public class DorisStreamLoad implements Serializable{
public void load(String value) throws StreamLoadException {
LOG.debug("Streamload Request:{} ,Body:{}", loadUrlStr, value);
LoadResponse loadResponse = loadBatch(value);
LOG.info("Streamload Response:{}",loadResponse);
if(loadResponse.status != 200){
throw new StreamLoadException("stream load error: " + loadResponse.respContent);
}else{
LOG.info("Streamload Response:{}",loadResponse);
ObjectMapper obj = new ObjectMapper();
try {
RespContent respContent = obj.readValue(loadResponse.respContent, RespContent.class);
@ -182,6 +182,7 @@ public class DorisStreamLoad implements Serializable{
HttpURLConnection feConn = null;
HttpURLConnection beConn = null;
int status = -1;
try {
// build request and send to new be location
beConn = getConnection(loadUrlStr, label);
@ -191,7 +192,7 @@ public class DorisStreamLoad implements Serializable{
bos.close();
// get respond
int status = beConn.getResponseCode();
status = beConn.getResponseCode();
String respMsg = beConn.getResponseMessage();
InputStream stream = (InputStream) beConn.getContent();
BufferedReader br = new BufferedReader(new InputStreamReader(stream));
@ -204,9 +205,9 @@ public class DorisStreamLoad implements Serializable{
} catch (Exception e) {
e.printStackTrace();
String err = "failed to execute spark streamload with label: " + label;
String err = "http request exception,load url : "+loadUrlStr+",failed to execute spark streamload with label: " + label;
LOG.warn(err, e);
return new LoadResponse(-1, e.getMessage(), err);
return new LoadResponse(status, e.getMessage(), err);
} finally {
if (feConn != null) {
feConn.disconnect();