Copilot commented on code in PR #3660:
URL: https://github.com/apache/hertzbeat/pull/3660#discussion_r2278707920
##########
hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/store/history/tsdb/vm/VictoriaMetricsDataStorage.java:
##########
@@ -591,36 +592,63 @@ private void
sendVictoriaMetrics(List<VictoriaMetricsDataStorage.VictoriaMetrics
}
}
// Refresh in advance to avoid waiting
- if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8) {
+ if (metricsBufferQueue.size() >= insertConfig.bufferSize() * 0.8
+ && draining.compareAndSet(false, true)) {
triggerImmediateFlush();
}
}
private void triggerImmediateFlush() {
- metricsFlushTimer.newTimeout(metricsFlushtask, 0,
TimeUnit.MILLISECONDS);
+ List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch = new
ArrayList<>(insertConfig.bufferSize());
+ metricsBufferQueue.drainTo(batch, insertConfig.bufferSize());
+ draining.set(false);
+ if (!batch.isEmpty()) {
+ metricsFlushTimer.newTimeout(new MetricsFlushTask(batch), 0,
TimeUnit.MILLISECONDS);
+ }
}
/**
* Regularly refresh the buffer queue to the vm
*/
private class MetricsFlushTask implements TimerTask {
+ private final List<VictoriaMetricsDataStorage.VictoriaMetricsContent>
batch;
+
+ public
MetricsFlushTask(List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch)
{
+ this.batch = batch;
+ }
+
@Override
public void run(Timeout timeout) {
try {
- List<VictoriaMetricsDataStorage.VictoriaMetricsContent> batch
= new ArrayList<>(insertConfig.bufferSize());
- metricsBufferQueue.drainTo(batch, insertConfig.bufferSize());
- if (!batch.isEmpty()) {
- doSaveData(batch);
- log.debug("[Victoria Metrics] Flushed {} metrics items",
batch.size());
- }
- if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
- metricsFlushTimer.newTimeout(this,
insertConfig.flushInterval(), TimeUnit.SECONDS);
- log.debug("[Victoria Metrics] Rescheduled next flush task
in {} seconds.", insertConfig.flushInterval());
+ if (batch == null) {
+ // If the batch is null, it means that the timer is
triggered by flush interval timer
+ List<VictoriaMetricsDataStorage.VictoriaMetricsContent>
batchT = new ArrayList<>(insertConfig.bufferSize());
+ metricsBufferQueue.drainTo(batchT,
insertConfig.bufferSize());
+ triggerDoSaveData(batchT);
+ // Reschedule the next flush task
+ triggerIntervalFlushTimer();
+ } else {
+ // If the batch is not null, it means that the timer is
triggered by the immediate flush
+ triggerDoSaveData(batch);
}
} catch (Exception e) {
log.error("[VictoriaMetrics] flush task error: {}",
e.getMessage(), e);
}
}
+
+ private void triggerDoSaveData(List<VictoriaMetricsContent> batch) {
+ if (!batch.isEmpty()) {
+ doSaveData(batch);
+ log.debug("[Victoria Metrics] Flushed {} metrics items",
batch.size());
+ }
+ }
+
+ private void triggerIntervalFlushTimer() {
+ if (metricsFlushTimer != null && !metricsFlushTimer.isStop()) {
+ metricsFlushTimer.newTimeout(this,
insertConfig.flushInterval(), TimeUnit.SECONDS);
Review Comment:
The method creates a new MetricsFlushTask instance by passing `this` to
`newTimeout()`, but it should pass `new MetricsFlushTask(null)` to maintain
consistency with the interval-based flush pattern established in line 134.
```suggestion
metricsFlushTimer.newTimeout(new MetricsFlushTask(null),
insertConfig.flushInterval(), TimeUnit.SECONDS);
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
