This is an automated email from the ASF dual-hosted git repository.
vernedeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 4152f98df9 [INLONG-10014][Audit] Add new audit SDK API for Sort Flink
4152f98df9 is described below
commit 4152f98df93a4869404ff3cb6f5310a974ca0a85
Author: doleyzi <[email protected]>
AuthorDate: Thu Apr 18 21:19:39 2024 +0800
[INLONG-10014][Audit] Add new audit SDK API for Sort Flink
---
.../org/apache/inlong/audit/AuditReporterImpl.java | 23 ++++++++++++++--------
1 file changed, 15 insertions(+), 8 deletions(-)
diff --git
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
index 48a24d1661..3474ac8e24 100644
---
a/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
+++
b/inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java
@@ -208,25 +208,32 @@ public class AuditReporterImpl implements Serializable {
/**
* Add audit data
*/
- public void add(int auditID, String inlongGroupID, String inlongStreamID,
Long logTime, long count, long size) {
+ public void add(int auditID, String inlongGroupID, String inlongStreamID,
long logTime, long count, long size) {
add(auditID, DEFAULT_AUDIT_TAG, inlongGroupID, inlongStreamID,
logTime, count, size, DEFAULT_AUDIT_VERSION);
}
- public void add(int auditID, String auditTag, String inlongGroupID, String
inlongStreamID, Long logTime,
+ public void add(long isolateKey, int auditID, String auditTag, String
inlongGroupID, String inlongStreamID,
+ long logTime, long count, long size, long auditVersion) {
+ long delayTime = System.currentTimeMillis() - logTime;
+ add(isolateKey, auditID, auditTag, inlongGroupID, inlongStreamID,
logTime,
+ count, size, delayTime, auditVersion);
+ }
+
+ public void add(int auditID, String auditTag, String inlongGroupID, String
inlongStreamID, long logTime,
long count, long size, long auditVersion) {
long delayTime = System.currentTimeMillis() - logTime;
- add(auditID, auditTag, inlongGroupID, inlongStreamID, logTime, count,
size,
+ add(DEFAULT_ISOLATE_KEY, auditID, auditTag, inlongGroupID,
inlongStreamID, logTime, count, size,
delayTime * count, auditVersion);
}
- public void add(int auditID, String inlongGroupID, String inlongStreamID,
Long logTime, long count, long size,
+ public void add(int auditID, String inlongGroupID, String inlongStreamID,
long logTime, long count, long size,
long delayTime) {
- add(auditID, DEFAULT_AUDIT_TAG, inlongGroupID, inlongStreamID,
logTime, count, size,
+ add(DEFAULT_ISOLATE_KEY, auditID, DEFAULT_AUDIT_TAG, inlongGroupID,
inlongStreamID, logTime, count, size,
delayTime, DEFAULT_AUDIT_VERSION);
}
- public void add(int auditID, String auditTag, String inlongGroupID, String
inlongStreamID, Long logTime,
- long count, long size, long delayTime, long auditVersion) {
+ public void add(long isolateKey, int auditID, String auditTag, String
inlongGroupID, String inlongStreamID,
+ long logTime, long count, long size, long delayTime, long
auditVersion) {
StringJoiner keyJoiner = new StringJoiner(FIELD_SEPARATORS);
keyJoiner.add(String.valueOf(logTime / PERIOD));
keyJoiner.add(inlongGroupID);
@@ -234,7 +241,7 @@ public class AuditReporterImpl implements Serializable {
keyJoiner.add(String.valueOf(auditID));
keyJoiner.add(auditTag);
keyJoiner.add(String.valueOf(auditVersion));
- addByKey(DEFAULT_ISOLATE_KEY, keyJoiner.toString(), count, size,
delayTime);
+ addByKey(isolateKey, keyJoiner.toString(), count, size, delayTime);
}
/**
