nzw921rx commented on code in PR #11778:
URL: https://github.com/apache/seatunnel/pull/11778#discussion_r3766115733
##########
seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriter.java:
##########
@@ -42,57 +44,44 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
@Slf4j
public class PrometheusWriter extends HttpSinkWriter {
private final List<Point> batchList;
- private volatile Exception flushException;
private final Integer batchSize;
- private final long flushInterval;
- private PrometheusSinkConfig sinkConfig;
+ private final PrometheusSinkConfig sinkConfig;
private final Serializer serializer;
protected final HttpClientProvider httpClient;
- private ScheduledExecutorService executor;
- private ScheduledFuture scheduledFuture;
public PrometheusWriter(
SeaTunnelRowType seaTunnelRowType,
HttpParameter httpParameter,
- ReadonlyConfig pluginConfig) {
+ ReadonlyConfig pluginConfig,
+ SinkWriter.Context context) {
super(seaTunnelRowType, httpParameter);
this.batchList = new ArrayList<>();
this.sinkConfig = PrometheusSinkConfig.loadConfig(pluginConfig);
this.batchSize = sinkConfig.getBatchSize();
- this.flushInterval = sinkConfig.getFlushInterval();
this.serializer =
new PrometheusSerializer(
seaTunnelRowType,
sinkConfig.getKeyTimestamp(),
sinkConfig.getKeyLabel(),
sinkConfig.getKeyValue());
this.httpClient = new HttpClientProvider(httpParameter);
- if (flushInterval > 0) {
- log.info("start schedule submit message,interval:{}",
flushInterval);
- this.executor =
- Executors.newScheduledThreadPool(
- 1,
- runnable -> {
- Thread thread = new Thread(runnable);
- thread.setDaemon(true);
- thread.setName("Prometheus-Metric-Sender");
- return thread;
- });
- this.scheduledFuture =
- executor.scheduleAtFixedRate(
- this::flushSchedule,
- flushInterval,
- flushInterval,
- TimeUnit.MILLISECONDS);
+ if
(pluginConfig.getOptional(PrometheusSinkOptions.FLUSH_INTERVAL).isPresent()) {
Review Comment:
The semantics of the actual parameters have been destructively replaced, and
I think we should clean up this parameter to avoid ambiguity🎉
--
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]
