You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "vernedeng (via GitHub)" <gi...@apache.org> on 2024/04/01 02:35:09 UTC
Re: [PR] [INLONG-9904][Audit] SDK support checkpoint feature [inlong]
vernedeng commented on code in PR #9905:
URL: https://github.com/apache/inlong/pull/9905#discussion_r1545903985
##########
inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java:
##########
@@ -54,19 +58,33 @@ public class AuditReporterImpl implements Serializable {
private static final int BATCH_NUM = 100;
private final ReentrantLock GLOBAL_LOCK = new ReentrantLock();
private static final int PERIOD = 1000 * 60;
- private final ConcurrentHashMap<String, StatInfo> countMap = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, StatInfo> threadCountMap = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, StatInfo> deleteCountMap = new ConcurrentHashMap<>();
- private final List<String> deleteKeyList = new ArrayList<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<String, StatInfo>> preStatMap =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<String, StatInfo>> summaryStatMap =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<String, StatInfo>> expiredStatMap =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, List<String>> expiredKeyList = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, Long> flushTime = new ConcurrentHashMap<>();
private final Config config = new Config();
private int packageId = 1;
private int dataId = 0;
private boolean initialized = false;
private SenderManager manager;
+ private AtomicInteger flushStat = new AtomicInteger(0);
private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
private AuditConfig auditConfig = null;
private SocketAddressListLoader loader = null;
+ private static final long DEFAULT_ISOLATE_KEY = 0;
+ private int flushStatThreshold = 100;
+ private boolean autoFlush = true;
+ public void setFlushStatThreshold(int flushStatThreshold) {
Review Comment:
Better add one blank line between variables and methods
##########
inlong-audit/audit-sdk/src/main/java/org/apache/inlong/audit/AuditReporterImpl.java:
##########
@@ -54,19 +58,33 @@ public class AuditReporterImpl implements Serializable {
private static final int BATCH_NUM = 100;
private final ReentrantLock GLOBAL_LOCK = new ReentrantLock();
private static final int PERIOD = 1000 * 60;
- private final ConcurrentHashMap<String, StatInfo> countMap = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, StatInfo> threadCountMap = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, StatInfo> deleteCountMap = new ConcurrentHashMap<>();
- private final List<String> deleteKeyList = new ArrayList<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<String, StatInfo>> preStatMap =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<String, StatInfo>> summaryStatMap =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, ConcurrentHashMap<String, StatInfo>> expiredStatMap =
+ new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, List<String>> expiredKeyList = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<Long, Long> flushTime = new ConcurrentHashMap<>();
private final Config config = new Config();
private int packageId = 1;
private int dataId = 0;
private boolean initialized = false;
private SenderManager manager;
+ private AtomicInteger flushStat = new AtomicInteger(0);
private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
private AuditConfig auditConfig = null;
private SocketAddressListLoader loader = null;
+ private static final long DEFAULT_ISOLATE_KEY = 0;
+ private int flushStatThreshold = 100;
+ private boolean autoFlush = true;
+ public void setFlushStatThreshold(int flushStatThreshold) {
+ this.flushStatThreshold = flushStatThreshold;
+ }
+ public void setAutoFlush(boolean autoFlush) {
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org