From 0773d5eee24856612bdf6505056ffa8c01d150db Mon Sep 17 00:00:00 2001 From: WenyXu Date: Thu, 17 Sep 2026 06:27:39 +0000 Subject: [PATCH] refactor: remove experimental prefix from pending rows batcher config Signed-off-by: WenyXu --- config/config.md | 32 +++++++++---------- config/frontend.example.toml | 4 +-- config/standalone.example.toml | 4 +-- src/frontend/src/frontend.rs | 29 +++++++----------- src/frontend/src/instance/builder.rs | 7 +++-- src/frontend/src/server.rs | 46 ++++++++++------------------ src/standalone/src/options.rs | 38 +++++++++-------------- 7 files changed, 65 insertions(+), 95 deletions(-) diff --git a/config/config.md b/config/config.md index 6325bb73e54..4dcffe40654 100644 --- a/config/config.md +++ b/config/config.md @@ -33,7 +33,7 @@ | `runtime.experimental_workload_scheduler.sample_every_polls` | Integer | `16` | Number of polls between scheduler fairness samples. Must be greater than zero. | | `http` | -- | -- | The HTTP server options. | | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. | -| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
raised to at least the largest active flush interval plus 1 second. The intervals come from
`prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`. | +| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
raised to at least the largest active flush interval plus 1 second. The intervals come from
`prom_store.pending_rows_flush_interval` and `pending_rows_batcher.pending_rows_flush_interval`. | | `http.body_limit` | String | `64MB` | HTTP request body limit.
The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
Set to 0 to disable limit. | | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
This allows browser to access http APIs without CORS restrictions | | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. | @@ -74,13 +74,13 @@ | `influxdb` | -- | -- | InfluxDB protocol options. | | `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. | | `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". | -| `experimental_pending_rows_batcher` | -- | -- | Shared experimental ordinary-table batching for opted-in ingestion protocols.
Legacy Prometheus batching settings under prom_store remain supported.
HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
Effective shared Prom settings take precedence; existing prom_store settings remain compatible. | -| `experimental_pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. | -| `experimental_pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. | -| `experimental_pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. | -| `experimental_pending_rows_batcher.worker_channel_capacity` | Integer | `65526` | Maximum queued submissions per table worker. | -| `experimental_pending_rows_batcher.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. | -| `experimental_pending_rows_batcher.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued table Flow notifications. | +| `pending_rows_batcher` | -- | -- | Shared experimental ordinary-table batching for opted-in ingestion protocols.
Legacy Prometheus batching settings under prom_store remain supported.
HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
Effective shared Prom settings take precedence; existing prom_store settings remain compatible. | +| `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. | +| `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. | +| `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. | +| `pending_rows_batcher.worker_channel_capacity` | Integer | `65526` | Maximum queued submissions per table worker. | +| `pending_rows_batcher.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. | +| `pending_rows_batcher.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued table Flow notifications. | | `jaeger` | -- | -- | Jaeger protocol options. | | `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. | | `otlp` | -- | -- | OpenTelemetry protocol options. | @@ -286,7 +286,7 @@ | `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.
Defaults to max(num_cpus / 2, 2). | | `http` | -- | -- | The HTTP server options. | | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. | -| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
raised to at least the largest active flush interval plus 1 second. The intervals come from
`prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`. | +| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
raised to at least the largest active flush interval plus 1 second. The intervals come from
`prom_store.pending_rows_flush_interval` and `pending_rows_batcher.pending_rows_flush_interval`. | | `http.body_limit` | String | `64MB` | HTTP request body limit.
The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
Set to 0 to disable limit. | | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
This allows browser to access http APIs without CORS restrictions | | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. | @@ -339,13 +339,13 @@ | `influxdb` | -- | -- | InfluxDB protocol options. | | `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. | | `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". | -| `experimental_pending_rows_batcher` | -- | -- | Shared experimental ordinary-table batching for opted-in ingestion protocols.
Legacy Prometheus batching settings under prom_store remain supported.
HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
Effective shared Prom settings take precedence; existing prom_store settings remain compatible. | -| `experimental_pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. | -| `experimental_pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. | -| `experimental_pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. | -| `experimental_pending_rows_batcher.worker_channel_capacity` | Integer | `65526` | Maximum queued submissions per table worker. | -| `experimental_pending_rows_batcher.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. | -| `experimental_pending_rows_batcher.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued table Flow notifications. | +| `pending_rows_batcher` | -- | -- | Shared experimental ordinary-table batching for opted-in ingestion protocols.
Legacy Prometheus batching settings under prom_store remain supported.
HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
Effective shared Prom settings take precedence; existing prom_store settings remain compatible. | +| `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. | +| `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. | +| `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. | +| `pending_rows_batcher.worker_channel_capacity` | Integer | `65526` | Maximum queued submissions per table worker. | +| `pending_rows_batcher.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. | +| `pending_rows_batcher.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued table Flow notifications. | | `jaeger` | -- | -- | Jaeger protocol options. | | `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. | | `otlp` | -- | -- | OpenTelemetry protocol options. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 4f70c86a040..95550931d46 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -54,7 +54,7 @@ addr = "127.0.0.1:4000" ## HTTP request timeout. Set to 0 to disable timeout. ## When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is ## raised to at least the largest active flush interval plus 1 second. The intervals come from -## `prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`. +## `prom_store.pending_rows_flush_interval` and `pending_rows_batcher.pending_rows_flush_interval`. timeout = "0s" ## HTTP request body limit. ## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`. @@ -233,7 +233,7 @@ default_merge_mode = "last_non_null" ## Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom. ## Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher. ## Effective shared Prom settings take precedence; existing prom_store settings remain compatible. -[experimental_pending_rows_batcher] +[pending_rows_batcher] # protocols = [ # "influxdb", # "opentsdb", diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 4c533a13611..54a1555232e 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -79,7 +79,7 @@ addr = "127.0.0.1:4000" ## HTTP request timeout. Set to 0 to disable timeout. ## When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is ## raised to at least the largest active flush interval plus 1 second. The intervals come from -## `prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`. +## `prom_store.pending_rows_flush_interval` and `pending_rows_batcher.pending_rows_flush_interval`. timeout = "0s" ## HTTP request body limit. ## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`. @@ -212,7 +212,7 @@ default_merge_mode = "last_non_null" ## Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom. ## Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher. ## Effective shared Prom settings take precedence; existing prom_store settings remain compatible. -[experimental_pending_rows_batcher] +[pending_rows_batcher] # protocols = [ # "influxdb", # "opentsdb", diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index b839493b127..793ab94205e 100644 --- a/src/frontend/src/frontend.rs +++ b/src/frontend/src/frontend.rs @@ -66,7 +66,7 @@ pub struct FrontendOptions { pub opentsdb: OpentsdbOptions, pub influxdb: InfluxdbOptions, /// Shared experimental ordinary-table batching; independent of Prom batching. - pub experimental_pending_rows_batcher: PendingRowsBatcherOptions, + pub pending_rows_batcher: PendingRowsBatcherOptions, pub prom_store: PromStoreOptions, pub jaeger: JaegerOptions, pub otlp: OtlpOptions, @@ -102,7 +102,7 @@ impl Default for FrontendOptions { postgres: PostgresOptions::default(), opentsdb: OpentsdbOptions::default(), influxdb: InfluxdbOptions::default(), - experimental_pending_rows_batcher: PendingRowsBatcherOptions::default(), + pending_rows_batcher: PendingRowsBatcherOptions::default(), jaeger: JaegerOptions::default(), prom_store: PromStoreOptions::default(), otlp: OtlpOptions::default(), @@ -127,7 +127,7 @@ impl Configurable for FrontendOptions { "heartbeat_env_vars", "meta_client.metasrv_addrs", "event_recorder.event_types", - "experimental_pending_rows_batcher.protocols", + "pending_rows_batcher.protocols", ]) } } @@ -226,14 +226,14 @@ mod tests { fn test_batcher_protocols_from_env() { temp_env::with_vars( [( - "FRONTEND_BATCHER_TEST__EXPERIMENTAL_PENDING_ROWS_BATCHER__PROTOCOLS", + "FRONTEND_BATCHER_TEST__PENDING_ROWS_BATCHER__PROTOCOLS", Some("influxdb,http_sql"), )], || { let options = FrontendOptions::load_layered_options(None, "FRONTEND_BATCHER_TEST").unwrap(); assert_eq!( - options.experimental_pending_rows_batcher.protocols, + options.pending_rows_batcher.protocols, vec![ servers::http::BatchingProtocol::Influxdb, servers::http::BatchingProtocol::HttpSql @@ -248,33 +248,26 @@ mod tests { let defaults: FrontendOptions = toml::from_str("").unwrap(); assert!( !defaults - .experimental_pending_rows_batcher + .pending_rows_batcher .pending_rows_batching_enabled() ); let options: FrontendOptions = toml::from_str( r#" -[experimental_pending_rows_batcher] +[pending_rows_batcher] protocols = ["influxdb", "http_sql"] pending_rows_flush_interval = "5ms" max_batch_rows = 25 "#, ) .unwrap(); - assert_eq!(options.experimental_pending_rows_batcher.max_batch_rows, 25); - assert_eq!(options.experimental_pending_rows_batcher.protocols.len(), 2); - assert!( - options - .experimental_pending_rows_batcher - .pending_rows_batching_enabled() - ); + assert_eq!(options.pending_rows_batcher.max_batch_rows, 25); + assert_eq!(options.pending_rows_batcher.protocols.len(), 2); + assert!(options.pending_rows_batcher.pending_rows_batching_enabled()); let serialized = toml::to_string(&options).unwrap(); let parsed: FrontendOptions = toml::from_str(&serialized).unwrap(); assert_eq!(options.influxdb, parsed.influxdb); assert_eq!(options.opentsdb, parsed.opentsdb); - assert_eq!( - options.experimental_pending_rows_batcher, - parsed.experimental_pending_rows_batcher - ); + assert_eq!(options.pending_rows_batcher, parsed.pending_rows_batcher); } #[test] diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index cce712b9e42..7a6ba474168 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -259,9 +259,10 @@ impl FrontendBuilder { ) .map(|batcher| batcher as Arc) }; - let inserter = Arc::new(create_inserter().with_pending_rows_batcher(build_batcher( - &self.options.experimental_pending_rows_batcher, - ))); + let inserter = Arc::new( + create_inserter() + .with_pending_rows_batcher(build_batcher(&self.options.pending_rows_batcher)), + ); let deleter = Arc::new(Deleter::new( self.catalog_manager.clone(), partition_manager.clone(), diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs index 9a454b1c202..257e5243e89 100644 --- a/src/frontend/src/server.rs +++ b/src/frontend/src/server.rs @@ -107,7 +107,7 @@ where request_memory_limiter: ServerMemoryLimiter, ) -> HttpServerBuilder { let mut builder = HttpServerBuilder::new(effective_http_options(opts)) - .with_batching_protocols(opts.experimental_pending_rows_batcher.protocols.clone()) + .with_batching_protocols(opts.pending_rows_batcher.protocols.clone()) .with_memory_limiter(request_memory_limiter) .with_sql_handler(self.instance.clone()); @@ -432,7 +432,7 @@ where /// Selected shared controls override legacy Prom batching knobs, not protocol behavior. fn effective_prom_store_options(opts: &FrontendOptions) -> PromStoreOptions { let mut prom_store = opts.prom_store.clone(); - let shared = &opts.experimental_pending_rows_batcher; + let shared = &opts.pending_rows_batcher; if shared.protocols.contains(&BatchingProtocol::Prom) && shared.pending_rows_batching_enabled() { prom_store.pending_rows_flush_interval = shared.pending_rows_flush_interval; @@ -452,7 +452,7 @@ fn effective_http_options(opts: &FrontendOptions) -> HttpOptions { fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> HttpOptions { let mut http = opts.http.clone(); let prom_store = effective_prom_store_options(opts); - let shared = &opts.experimental_pending_rows_batcher; + let shared = &opts.pending_rows_batcher; // Ordinary-table batching always waits for its flush, independently of the // dedicated Prom batcher's asynchronous acknowledgement mode. let common_enabled = shared.pending_rows_batching_enabled() @@ -525,7 +525,7 @@ mod tests { opts.prom_store.enable = prom_enabled; opts.prom_store .experimental_enable_prometheus_native_histogram = true; - let shared = &mut opts.experimental_pending_rows_batcher; + let shared = &mut opts.pending_rows_batcher; shared.protocols = vec![if selected { BatchingProtocol::Prom } else { @@ -588,9 +588,9 @@ mod tests { opts.http.timeout = Duration::from_secs(timeout_secs); opts.prom_store.with_metric_engine = metric_engine; opts.prom_store.pending_rows_flush_interval = Duration::from_secs(legacy_secs); - opts.experimental_pending_rows_batcher.protocols = protocols; - opts.experimental_pending_rows_batcher - .pending_rows_flush_interval = Duration::from_secs(shared_secs); + opts.pending_rows_batcher.protocols = protocols; + opts.pending_rows_batcher.pending_rows_flush_interval = + Duration::from_secs(shared_secs); assert_eq!( effective_http_options_with_sync(&opts, batch_sync).timeout, Duration::from_secs(expected_secs) @@ -682,38 +682,24 @@ mod tests { fn test_invalid_shared_batching_preserves_prom_store_options() { type KnobMutator = fn(&mut FrontendOptions); let cases: [KnobMutator; 5] = [ + |opts| opts.pending_rows_batcher.max_concurrent_flushes = usize::MAX, + |opts| opts.pending_rows_batcher.worker_channel_capacity = usize::MAX, + |opts| opts.pending_rows_batcher.max_inflight_requests = usize::MAX, |opts| { - opts.experimental_pending_rows_batcher - .max_concurrent_flushes = usize::MAX - }, - |opts| { - opts.experimental_pending_rows_batcher - .worker_channel_capacity = usize::MAX - }, - |opts| opts.experimental_pending_rows_batcher.max_inflight_requests = usize::MAX, - |opts| { - opts.experimental_pending_rows_batcher - .flow_notification_queue_capacity = NonZeroUsize::new(usize::MAX).unwrap() - }, - |opts| { - opts.experimental_pending_rows_batcher - .pending_rows_flush_interval = Duration::MAX + opts.pending_rows_batcher.flow_notification_queue_capacity = + NonZeroUsize::new(usize::MAX).unwrap() }, + |opts| opts.pending_rows_batcher.pending_rows_flush_interval = Duration::MAX, ]; for invalidate in cases { let mut opts = FrontendOptions::default(); opts.http.timeout = Duration::from_secs(1); opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5); - opts.experimental_pending_rows_batcher.protocols = + opts.pending_rows_batcher.protocols = vec![BatchingProtocol::Prom, BatchingProtocol::Influxdb]; - opts.experimental_pending_rows_batcher - .pending_rows_flush_interval = Duration::from_secs(10); + opts.pending_rows_batcher.pending_rows_flush_interval = Duration::from_secs(10); invalidate(&mut opts); - assert!( - !opts - .experimental_pending_rows_batcher - .pending_rows_batching_enabled() - ); + assert!(!opts.pending_rows_batcher.pending_rows_batching_enabled()); assert_eq!(opts.prom_store, effective_prom_store_options(&opts)); assert_eq!( Duration::from_secs(6), diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index c6ed74c6f74..4884b5ee091 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -57,7 +57,7 @@ pub struct StandaloneOptions { pub opentsdb: OpentsdbOptions, pub influxdb: InfluxdbOptions, /// Shared experimental ordinary-table batching; independent of Prom batching. - pub experimental_pending_rows_batcher: PendingRowsBatcherOptions, + pub pending_rows_batcher: PendingRowsBatcherOptions, pub jaeger: JaegerOptions, pub otlp: OtlpOptions, pub prom_store: PromStoreOptions, @@ -99,7 +99,7 @@ impl Default for StandaloneOptions { postgres: PostgresOptions::default(), opentsdb: OpentsdbOptions::default(), influxdb: InfluxdbOptions::default(), - experimental_pending_rows_batcher: PendingRowsBatcherOptions::default(), + pending_rows_batcher: PendingRowsBatcherOptions::default(), jaeger: JaegerOptions::default(), otlp: OtlpOptions::default(), prom_store: PromStoreOptions::default(), @@ -133,7 +133,7 @@ impl Configurable for StandaloneOptions { "heartbeat_env_vars", "wal.broker_endpoints", "event_recorder.event_types", - "experimental_pending_rows_batcher.protocols", + "pending_rows_batcher.protocols", ]) } } @@ -162,7 +162,7 @@ impl StandaloneOptions { postgres: cloned_opts.postgres, opentsdb: cloned_opts.opentsdb, influxdb: cloned_opts.influxdb, - experimental_pending_rows_batcher: cloned_opts.experimental_pending_rows_batcher, + pending_rows_batcher: cloned_opts.pending_rows_batcher, jaeger: cloned_opts.jaeger, otlp: cloned_opts.otlp, prom_store: cloned_opts.prom_store, @@ -219,7 +219,7 @@ mod tests { fn test_batcher_protocols_from_env() { temp_env::with_vars( [( - "STANDALONE_BATCHER_TEST__EXPERIMENTAL_PENDING_ROWS_BATCHER__PROTOCOLS", + "STANDALONE_BATCHER_TEST__PENDING_ROWS_BATCHER__PROTOCOLS", Some("influxdb,http_sql"), )], || { @@ -227,7 +227,7 @@ mod tests { StandaloneOptions::load_layered_options(None, "STANDALONE_BATCHER_TEST") .unwrap(); assert_eq!( - options.experimental_pending_rows_batcher.protocols, + options.pending_rows_batcher.protocols, vec![ servers::http::BatchingProtocol::Influxdb, servers::http::BatchingProtocol::HttpSql @@ -242,12 +242,12 @@ mod tests { let defaults: StandaloneOptions = toml::from_str("").unwrap(); assert!( !defaults - .experimental_pending_rows_batcher + .pending_rows_batcher .pending_rows_batching_enabled() ); let options: StandaloneOptions = toml::from_str( r#" -[experimental_pending_rows_batcher] +[pending_rows_batcher] protocols = ["influxdb", "http_sql"] pending_rows_flush_interval = "5ms" max_batch_rows = 25 @@ -255,35 +255,25 @@ flow_notification_queue_capacity = 17 "#, ) .unwrap(); - assert_eq!(options.experimental_pending_rows_batcher.max_batch_rows, 25); - assert_eq!(options.experimental_pending_rows_batcher.protocols.len(), 2); + assert_eq!(options.pending_rows_batcher.max_batch_rows, 25); + assert_eq!(options.pending_rows_batcher.protocols.len(), 2); assert_eq!( options - .experimental_pending_rows_batcher + .pending_rows_batcher .flow_notification_queue_capacity .get(), 17 ); - assert!( - options - .experimental_pending_rows_batcher - .pending_rows_batching_enabled() - ); + assert!(options.pending_rows_batcher.pending_rows_batching_enabled()); let serialized = toml::to_string(&options).unwrap(); let parsed: StandaloneOptions = toml::from_str(&serialized).unwrap(); assert_eq!(options.influxdb, parsed.influxdb); assert_eq!(options.opentsdb, parsed.opentsdb); - assert_eq!( - options.experimental_pending_rows_batcher, - parsed.experimental_pending_rows_batcher - ); + assert_eq!(options.pending_rows_batcher, parsed.pending_rows_batcher); let frontend = options.frontend_options(); assert_eq!(options.influxdb, frontend.influxdb); assert_eq!(options.opentsdb, frontend.opentsdb); - assert_eq!( - options.experimental_pending_rows_batcher, - frontend.experimental_pending_rows_batcher - ); + assert_eq!(options.pending_rows_batcher, frontend.pending_rows_batcher); } #[test]