hubcio commented on code in PR #3377:
URL: https://github.com/apache/iggy/pull/3377#discussion_r3340715640
##########
core/integration/tests/connectors/random/random_source.rs:
##########
@@ -17,79 +17,33 @@
* under the License.
*/
-use iggy_common::MessageClient;
-use iggy_common::{Consumer, Identifier, PollingStrategy};
+use crate::connectors::source_suite;
use integration::harness::seeds;
use integration::iggy_harness;
-use std::time::Duration;
-use tokio::time::sleep;
#[iggy_harness(
server(connectors_runtime(config_path =
"tests/connectors/random/source.toml")),
seed = seeds::connector_stream
)]
async fn random_source_produces_messages(harness: &TestHarness) {
- sleep(Duration::from_secs(1)).await;
-
- let client = harness.root_client().await.unwrap();
- let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
- let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
- let consumer_id: Identifier = "test_consumer".try_into().unwrap();
-
- let messages = client
- .poll_messages(
- &stream_id,
- &topic_id,
- None,
- &Consumer::new(consumer_id),
- &PollingStrategy::next(),
- 10,
- true,
- )
- .await
- .expect("Failed to poll messages");
-
+ let messages =
source_suite::assert_source_produces_messages(harness).await;
assert!(
- !messages.messages.is_empty(),
+ !messages.is_empty(),
"No messages received from random source"
);
- assert!(
- messages.current_offset > 0,
- "Current offset should be greater than 0"
- );
}
#[iggy_harness(
server(connectors_runtime(config_path =
"tests/connectors/random/source.toml")),
seed = seeds::connector_stream
)]
async fn state_persists_across_connector_restart(harness: &mut TestHarness) {
Review Comment:
this test can't fail. it polls two different fresh consumers
(`source_suite_before_restart`, then `source_suite_after_restart`) against a
source that emits forever, and only checks `!after.is_empty()`. a fresh
consumer gets new messages whether or not connector state survived the restart,
so a state-loss regression still passes green - which is worse than no test
under this name.
the old `current_offset` check was the partition head too (it's documented
as the partition's offset, not the consumer's), so it never proved cursor
survival either. the pattern to copy is the postgres source restart test from
#2579: one consumer across the restart, seed a second batch with
strictly-greater ids, assert every post-restart id is greater than the
pre-restart max. random can't supply that discriminator, so either move this
test to postgres or drop it and keep random as a liveness smoke (and rename it).
##########
core/integration/tests/connectors/random/random_source.rs:
##########
@@ -17,79 +17,33 @@
* under the License.
*/
-use iggy_common::MessageClient;
-use iggy_common::{Consumer, Identifier, PollingStrategy};
+use crate::connectors::source_suite;
use integration::harness::seeds;
use integration::iggy_harness;
-use std::time::Duration;
-use tokio::time::sleep;
#[iggy_harness(
server(connectors_runtime(config_path =
"tests/connectors/random/source.toml")),
seed = seeds::connector_stream
)]
async fn random_source_produces_messages(harness: &TestHarness) {
- sleep(Duration::from_secs(1)).await;
-
- let client = harness.root_client().await.unwrap();
- let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
- let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
- let consumer_id: Identifier = "test_consumer".try_into().unwrap();
-
- let messages = client
- .poll_messages(
- &stream_id,
- &topic_id,
- None,
- &Consumer::new(consumer_id),
- &PollingStrategy::next(),
- 10,
- true,
- )
- .await
- .expect("Failed to poll messages");
-
+ let messages =
source_suite::assert_source_produces_messages(harness).await;
assert!(
Review Comment:
redundant - `assert_source_produces_messages` already asserts non-empty
before returning. drop this, or call `poll_until_min_messages` directly if you
want the explicit check here.
##########
core/integration/tests/connectors/source_suite.rs:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::time::Duration;
+
+use iggy_common::{Consumer, Identifier, IggyMessage, MessageClient,
PollingStrategy};
+use integration::harness::{TestHarness, seeds};
+use tokio::time::{sleep, timeout};
+
+pub(crate) struct SourceSuiteConfig {
+ pub(crate) consumer_name: &'static str,
+ pub(crate) min_messages: usize,
+ pub(crate) poll_batch: u32,
+ pub(crate) warmup: Duration,
+ pub(crate) retry_interval: Duration,
+ pub(crate) timeout: Duration,
+}
+
+impl Default for SourceSuiteConfig {
+ fn default() -> Self {
+ Self {
+ consumer_name: "source_suite_consumer",
+ min_messages: 1,
+ poll_batch: 100,
+ warmup: Duration::from_secs(1),
+ retry_interval: Duration::from_millis(100),
+ timeout: Duration::from_secs(5),
+ }
+ }
+}
+
+pub(crate) async fn poll_until_min_messages(
+ harness: &TestHarness,
+ config: &SourceSuiteConfig,
+) -> Vec<IggyMessage> {
+ sleep(config.warmup).await;
+
+ let client = harness.root_client().await.expect("root client");
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = config.consumer_name.try_into().unwrap();
+
+ let poll = async {
+ loop {
+ let polled = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ config.poll_batch,
+ true,
+ )
+ .await
+ .expect("poll source messages");
+
+ if polled.messages.len() >= config.min_messages {
Review Comment:
this returns a single poll iteration's messages, but with `auto_commit=true`
above each poll commits its batch. when `min_messages` is larger than one
server batch, the earlier batches are committed and then dropped here - never
accumulated - so the loop can spin to the timeout while silently discarding
everything it read. latent now since the default `min_messages` is 1, but it
bites the moment a real source is wired with a higher count. accumulate into a
`Vec` across iterations like the postgres and influxdb source loops do.
##########
core/integration/tests/connectors/source_suite.rs:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::time::Duration;
+
+use iggy_common::{Consumer, Identifier, IggyMessage, MessageClient,
PollingStrategy};
+use integration::harness::{TestHarness, seeds};
+use tokio::time::{sleep, timeout};
+
+pub(crate) struct SourceSuiteConfig {
+ pub(crate) consumer_name: &'static str,
+ pub(crate) min_messages: usize,
+ pub(crate) poll_batch: u32,
+ pub(crate) warmup: Duration,
+ pub(crate) retry_interval: Duration,
+ pub(crate) timeout: Duration,
+}
+
+impl Default for SourceSuiteConfig {
+ fn default() -> Self {
+ Self {
+ consumer_name: "source_suite_consumer",
+ min_messages: 1,
+ poll_batch: 100,
+ warmup: Duration::from_secs(1),
+ retry_interval: Duration::from_millis(100),
+ timeout: Duration::from_secs(5),
+ }
+ }
+}
+
+pub(crate) async fn poll_until_min_messages(
+ harness: &TestHarness,
+ config: &SourceSuiteConfig,
+) -> Vec<IggyMessage> {
+ sleep(config.warmup).await;
+
+ let client = harness.root_client().await.expect("root client");
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = config.consumer_name.try_into().unwrap();
+
+ let poll = async {
+ loop {
+ let polled = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ config.poll_batch,
+ true,
+ )
+ .await
+ .expect("poll source messages");
+
+ if polled.messages.len() >= config.min_messages {
+ return polled.messages;
+ }
+
+ sleep(config.retry_interval).await;
+ }
+ };
+
+ timeout(config.timeout, poll)
+ .await
+ .expect("source suite timed out waiting for messages")
Review Comment:
on timeout this doesn't say which consumer or config timed out - include
`consumer_name` so a failure points at the right test.
##########
core/integration/tests/connectors/source_suite.rs:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::time::Duration;
+
+use iggy_common::{Consumer, Identifier, IggyMessage, MessageClient,
PollingStrategy};
+use integration::harness::{TestHarness, seeds};
+use tokio::time::{sleep, timeout};
+
+pub(crate) struct SourceSuiteConfig {
+ pub(crate) consumer_name: &'static str,
+ pub(crate) min_messages: usize,
+ pub(crate) poll_batch: u32,
+ pub(crate) warmup: Duration,
+ pub(crate) retry_interval: Duration,
+ pub(crate) timeout: Duration,
+}
+
+impl Default for SourceSuiteConfig {
+ fn default() -> Self {
+ Self {
+ consumer_name: "source_suite_consumer",
+ min_messages: 1,
+ poll_batch: 100,
+ warmup: Duration::from_secs(1),
+ retry_interval: Duration::from_millis(100),
+ timeout: Duration::from_secs(5),
+ }
+ }
+}
+
+pub(crate) async fn poll_until_min_messages(
+ harness: &TestHarness,
+ config: &SourceSuiteConfig,
+) -> Vec<IggyMessage> {
+ sleep(config.warmup).await;
Review Comment:
this fixed 1s sleep is dead time - it's before and outside the `timeout`
below, so worst case is warmup + timeout. the harness already gates runtime
readiness with a health-poll on startup, and the retry loop below covers
not-ready, which is why the existing postgres/elasticsearch/influxdb source
loops poll immediately with no warmup. drop it.
##########
core/integration/tests/connectors/source_suite.rs:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::time::Duration;
+
+use iggy_common::{Consumer, Identifier, IggyMessage, MessageClient,
PollingStrategy};
+use integration::harness::{TestHarness, seeds};
+use tokio::time::{sleep, timeout};
+
+pub(crate) struct SourceSuiteConfig {
+ pub(crate) consumer_name: &'static str,
+ pub(crate) min_messages: usize,
+ pub(crate) poll_batch: u32,
+ pub(crate) warmup: Duration,
Review Comment:
this hardcodes its own retry budget (`warmup`/`retry_interval`/`timeout`).
worth noting there isn't a single shared constant to reuse yet - postgres and
elasticsearch each carry their own `POLL_ATTEMPTS=100`/`POLL_INTERVAL_MS=50`
copy and influxdb just hardcodes `0..100`/`from_millis(100)`. a helper whose
point is to absorb duplication should unify all of those into one place rather
than add a fourth timing style.
##########
core/integration/tests/connectors/source_suite.rs:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::time::Duration;
+
+use iggy_common::{Consumer, Identifier, IggyMessage, MessageClient,
PollingStrategy};
+use integration::harness::{TestHarness, seeds};
+use tokio::time::{sleep, timeout};
+
+pub(crate) struct SourceSuiteConfig {
Review Comment:
as-is this can only assert liveness - there's no notion of message identity,
exact count, ordering, or the connector cursor, so it can't host the checks
#2892 is about (restart-resume needs an id discriminator, replay/dup needs `==
N`). if this is meant to grow into the suite, the shape wants to be a
fixture/trait the runner is generic over (a source-under-test that seeds n
stamped records and states its expected set), not a free-function poll. sketch
is in the review summary.
##########
core/integration/tests/connectors/source_suite.rs:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use std::time::Duration;
+
+use iggy_common::{Consumer, Identifier, IggyMessage, MessageClient,
PollingStrategy};
+use integration::harness::{TestHarness, seeds};
+use tokio::time::{sleep, timeout};
+
+pub(crate) struct SourceSuiteConfig {
+ pub(crate) consumer_name: &'static str,
+ pub(crate) min_messages: usize,
+ pub(crate) poll_batch: u32,
+ pub(crate) warmup: Duration,
+ pub(crate) retry_interval: Duration,
+ pub(crate) timeout: Duration,
+}
+
+impl Default for SourceSuiteConfig {
+ fn default() -> Self {
+ Self {
+ consumer_name: "source_suite_consumer",
+ min_messages: 1,
+ poll_batch: 100,
+ warmup: Duration::from_secs(1),
+ retry_interval: Duration::from_millis(100),
+ timeout: Duration::from_secs(5),
+ }
+ }
+}
+
+pub(crate) async fn poll_until_min_messages(
+ harness: &TestHarness,
+ config: &SourceSuiteConfig,
+) -> Vec<IggyMessage> {
+ sleep(config.warmup).await;
+
+ let client = harness.root_client().await.expect("root client");
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+ let consumer_id: Identifier = config.consumer_name.try_into().unwrap();
+
+ let poll = async {
+ loop {
+ let polled = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &Consumer::new(consumer_id.clone()),
+ &PollingStrategy::next(),
+ config.poll_batch,
+ true,
+ )
+ .await
+ .expect("poll source messages");
Review Comment:
`.expect` here panics the test on any transient poll error instead of
retrying. the postgres and elasticsearch source loops use `if let Ok(...)` and
keep looping until their timeout, which is the more robust pattern. (the
influxdb source loops already use `.expect` too, so they share this fragility -
worth fixing there rather than copying it here.)
--
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]
