surafel58 commented on code in PR #11778:
URL: https://github.com/apache/seatunnel/pull/11778#discussion_r3766179822
##########
docs/zh/connectors/sink/Prometheus.md:
##########
@@ -51,7 +51,7 @@ Prometheus 数据接收器把上游数据写入 Prometheus remote write API。
| retry_backoff_multiplier_ms | Int | 否 | 100 | 重试退避时间倍数，单位毫秒。 |
| retry_backoff_max_ms | Int | 否 | 10000 | 最大重试退避时间，单位毫秒。 |
| batch_size | Int | 否 | 1024 | 写入 Prometheus
前最多缓存的行数。 |
-| flush_interval | Long | 否 | 300000 | 最大刷新间隔，单位毫秒。 |
+| flush_interval | Long | 否 | 300000 |
已废弃且不再生效。连接器不再启动自己的刷新线程。请改为在作业的 `env` 中设置引擎级的
`sink.flush.interval`（详见[定时刷新](#定时刷新)）。 |
Review Comment:
Got it, I will remove this parameter, add an entry for it in
incompatible-changes.md, and keep the English and Chinese docs in sync.
##########
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:
On it, I will remove the flush_interval option completely from
PrometheusWriter, PrometheusSinkConfig, PrometheusSinkOptions, and the option
rule, so the parameter is no longer there.
--
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]
