[INLONG-9739][Sort] onParsingBodyFailure function in FailureHandler c…

…lass need add 'head' paramete to generate failure metric (#9740)
apache · Feb 28, 2024 · a94f0fa · a94f0fa
1 parent 9a9a436
commit a94f0fa
Show file tree

Hide file tree

Showing 12 changed files with 14 additions and 14 deletions.
diff --git a/.../format-common/src/main/java/org/apache/inlong/sort/formats/inlongmsg/FailureHandler.java b/.../format-common/src/main/java/org/apache/inlong/sort/formats/inlongmsg/FailureHandler.java
@@ -40,7 +40,7 @@ public interface FailureHandler extends Serializable {
  * @param exception the thrown exception
  * @throws Exception the exception
  */
- void onParsingBodyFailure(byte[] body, Exception exception) throws Exception;
+ void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception;
 
  /**
  * This method is called when there is a failure occurred while converting head and body to row.

diff --git a/...t-common/src/main/java/org/apache/inlong/sort/formats/inlongmsg/IgnoreFailureHandler.java b/...t-common/src/main/java/org/apache/inlong/sort/formats/inlongmsg/IgnoreFailureHandler.java
@@ -33,8 +33,8 @@ public void onParsingHeadFailure(String attribute, Exception exception) {
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) {
- LOG.warn("Cannot properly parse the body: {}.", new String(body), exception);
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) {
+ LOG.warn("Cannot properly parse the head: {}, body: {}.", head, new String(body), exception);
  }
 
  @Override

diff --git a/...mat-common/src/main/java/org/apache/inlong/sort/formats/inlongmsg/NoOpFailureHandler.java b/...mat-common/src/main/java/org/apache/inlong/sort/formats/inlongmsg/NoOpFailureHandler.java
@@ -34,8 +34,8 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
- LOG.error("Cannot properly parse the body: {}.", new String(body), exception);
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
+ LOG.error("Cannot properly parse the head: {}, the body: {}.", head, new String(body), exception);
  throw exception;
  }
 

diff --git a/...in/java/org/apache/inlong/sort/formats/inlongmsg/AbstractInLongMsgFormatDeserializer.java b/...in/java/org/apache/inlong/sort/formats/inlongmsg/AbstractInLongMsgFormatDeserializer.java
@@ -103,7 +103,7 @@ public void flatMap(
  try {
  bodyList = parseBodyList(bodyBytes);
  } catch (Exception e) {
- failureHandler.onParsingBodyFailure(bodyBytes, e);
+ failureHandler.onParsingBodyFailure(head, bodyBytes, e);
  continue;
  }
 

diff --git a/.../java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializerTest.java b/.../java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializerTest.java
@@ -725,7 +725,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }
 

diff --git a/...st/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvFormatDeserializerTest.java b/...st/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvFormatDeserializerTest.java
@@ -319,7 +319,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }
 

diff --git a/...g/apache/inlong/sort/formats/inlongmsgtlogcsv/InLongMsgTlogCsvFormatDeserializerTest.java b/...g/apache/inlong/sort/formats/inlongmsgtlogcsv/InLongMsgTlogCsvFormatDeserializerTest.java
@@ -272,7 +272,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }
 

diff --git a/...org/apache/inlong/sort/formats/inlongmsgtlogkv/InLongMsgTlogKvFormatDeserializerTest.java b/...org/apache/inlong/sort/formats/inlongmsgtlogkv/InLongMsgTlogKvFormatDeserializerTest.java
@@ -200,7 +200,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }
 

diff --git a/...in/java/org/apache/inlong/sort/formats/inlongmsg/AbstractInLongMsgFormatDeserializer.java b/...in/java/org/apache/inlong/sort/formats/inlongmsg/AbstractInLongMsgFormatDeserializer.java
@@ -111,7 +111,7 @@ public List<InLongMsgWrap> preParse(byte[] bytes) throws Exception {
  bodyList = parseBodyList(bodyBytes);
  } catch (Exception e) {
  reportDeSerializeErrorMetrics();
- failureHandler.onParsingBodyFailure(bodyBytes, e);
+ failureHandler.onParsingBodyFailure(head, bodyBytes, e);
  continue;
  }
 

diff --git a/.../java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializerTest.java b/.../java/org/apache/inlong/sort/formats/inlongmsgcsv/InLongMsgCsvFormatDeserializerTest.java
@@ -848,7 +848,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }
 

diff --git a/...st/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvFormatDeserializerTest.java b/...st/java/org/apache/inlong/sort/formats/inlongmsgkv/InLongMsgKvFormatDeserializerTest.java
@@ -332,7 +332,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }
 

diff --git a/...g/apache/inlong/sort/formats/inlongmsgtlogcsv/InLongMsgTlogCsvFormatDeserializerTest.java b/...g/apache/inlong/sort/formats/inlongmsgtlogcsv/InLongMsgTlogCsvFormatDeserializerTest.java
@@ -324,7 +324,7 @@ public void onParsingHeadFailure(String attribute, Exception exception) throws E
  }
 
  @Override
- public void onParsingBodyFailure(byte[] body, Exception exception) throws Exception {
+ public void onParsingBodyFailure(InLongMsgHead head, byte[] body, Exception exception) throws Exception {
  bodyCount++;
  }