|
|
@@ -1,10 +1,12 @@
|
|
|
package cn.com.ctop.job.bytedance.utils.doris.sink;
|
|
|
|
|
|
import cn.com.ctop.job.bytedance.data.utils.JsonUtils;
|
|
|
+import com.alibaba.fastjson.JSONArray;
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.apache.http.HttpEntity;
|
|
|
import org.apache.http.client.methods.CloseableHttpResponse;
|
|
|
+import org.apache.http.client.methods.HttpPost;
|
|
|
import org.apache.http.client.methods.HttpPut;
|
|
|
import org.apache.http.entity.StringEntity;
|
|
|
import org.apache.http.impl.client.CloseableHttpClient;
|
|
|
@@ -75,6 +77,7 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
|
|
|
|
|
|
put.setEntity(new StringEntity(jsonBody, StandardCharsets.UTF_8));
|
|
|
|
|
|
+
|
|
|
try (CloseableHttpClient httpClient = HttpClients.custom()
|
|
|
.setRedirectStrategy(new StrictRedirectStrategy()) // 支持 307/301 重定向
|
|
|
.build();
|
|
|
@@ -82,14 +85,37 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
|
|
|
int code = response.getStatusLine().getStatusCode();
|
|
|
HttpEntity entity = response.getEntity();
|
|
|
String responseBody = EntityUtils.toString(entity, "UTF-8");
|
|
|
+ String fsWebhookUrl = "https://open.feishu.cn/open-apis/bot/v2/hook/371e7945-7258-4574-a2eb-d7cd0490f86c";
|
|
|
if (code != 200 && code != 201) {
|
|
|
+ sendFsMessage(fsWebhookUrl, "stream load 异常 " + label, "Doris stream load failed. Status: " + code + ", response: " + responseBody);
|
|
|
throw new RuntimeException("Doris stream load failed. Status: " + code + ", response: " + responseBody);
|
|
|
}
|
|
|
// 可选: 解析返回内容判断 success
|
|
|
- if (!responseBody.contains("\"Status\": \"Success\"")) {
|
|
|
+ if (!responseBody.contains("\"Status\": \"Success\"") || responseBody.contains("ErrorURL")) {
|
|
|
+ sendFsMessage(fsWebhookUrl, "stream load 异常 " + label, responseBody);
|
|
|
throw new RuntimeException("Stream load returned non-success: " + responseBody);
|
|
|
}
|
|
|
log.info("Doris stream load success: {}", responseBody);
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
+ //发送飞书消息
|
|
|
+ private static void sendFsMessage(String url, String head, String message) {
|
|
|
+ JSONObject body = new JSONObject();
|
|
|
+ body.put("msg_type", "interactive");
|
|
|
+ JSONArray elements = new JSONArray();
|
|
|
+ elements.add(new JSONObject()
|
|
|
+ .fluentPut("tag", "div")
|
|
|
+ .fluentPut("text", new JSONObject().fluentPut("content", message).fluentPut("tag", "lark_md")));
|
|
|
+ JSONObject header = new JSONObject().fluentPut("title", new JSONObject().fluentPut("content", head).fluentPut("tag", "plain_text"));
|
|
|
+ body.put("card", new JSONObject().fluentPut("elements", elements).fluentPut("header", header));
|
|
|
+ HttpPost httpPost = new HttpPost(url);
|
|
|
+ httpPost.setEntity(new StringEntity(body.toString(), StandardCharsets.UTF_8));
|
|
|
+ try (CloseableHttpClient httpClient = HttpClients.custom().build();
|
|
|
+ CloseableHttpResponse response = httpClient.execute(httpPost)) {
|
|
|
+ response.getStatusLine().getStatusCode();
|
|
|
+ } catch (Exception e) {
|
|
|
+ e.printStackTrace();
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|