zhangshenghang commented on code in PR #10925:
URL: https://github.com/apache/seatunnel/pull/10925#discussion_r3675121361
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/resources/kafka/kafka_to_kafka_exactly_once_streaming.conf:
##########
@@ -24,11 +24,13 @@ env {
source {
Kafka {
bootstrap.servers = "kafkaCluster:9092"
- topic = "${sourceTopic}"
- consumer.group = "${consumerGroup}"
+ topic = "${sourceTopic:kafka_topic_exactly_once_1}"
+ consumer.group = "${consumerGroup:test_exactly_once}"
# The default format is json, which is optional
format = text
- start_mode = group_offsets
+ # The exactly-once test sends a readiness probe while the job starts.
+ # Read from earliest so a probe sent before split assignment is not
skipped.
+ start_mode = earliest
Review Comment:
**P1 — Why `earliest` instead of `group_offsets`?**
This change is functionally critical (without it
`KafkaSourceSplitEnumerator.listConsumerGroupOffsets` returns an empty map for
a brand-new consumer group, so no split is assigned and the source never
reads). The inline comment is fine, but:
- Please also document the rationale in the PR description so reviewers
don't have to trace through `KafkaSourceSplitEnumerator` to understand it.
- The same conf file is used by
`testRestoreKafkaToKafkaExactlyOnceOnStreaming` (KafkaIT.java:1843). On
restore, `KafkaSourceSplitEnumerator` uses restored splits and `start_mode`
should not affect that path, but please add a short comment here so future
maintainers don't accidentally revert this when refactoring restore.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java:
##########
@@ -53,6 +54,12 @@ public class TDengineIT extends TestSuiteBase implements
TestResource {
private static final String NETWORK_ALIASES1 = "flink_e2e_tdengine_src";
private static final String NETWORK_ALIASES2 = "flink_e2e_tdengine_sink";
private static final int PORT = 6041;
+ /**
+ * Keeps the test database within the vnode budget that small CI machines
can provide.
+ *
+ * <p>A single vgroup is sufficient for connector assertions and avoids CI
startup failures.
+ */
+ private static final String SINGLE_VGROUP_DATABASE_OPTIONS = " KEEP 3650
VGROUPS 1";
Review Comment:
**P1 — `VGROUPS 1` may diverge from default CI/prod behavior**
Forcing every database to a single vgroup is a workaround for small CI
runners, but please confirm that none of the tested STABLE-backed source/sink
queries (`power.meters`, `power2.meters2/3/4`, `power3.meters5`) depend on
multi-vgroup parallelism or behavior. If the connector internally parallelizes
per-vgroup, a single-vgroup layout could change timing in CI in ways that mask
real regressions.
Suggested: add a comment that explicitly says "single vgroup is purely a CI
resource workaround and is expected to behave identically to multi-vgroup for
the queries exercised by this test."
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java:
##########
@@ -224,6 +203,37 @@ private Connection createConnect(GenericContainer<?>
tdengineServer) {
return conn;
}
+ /**
+ * Waits until TDengine accepts JDBC traffic and can allocate vgroups for
new databases.
+ *
+ * <p>Port readiness alone does not prove that TDengine can create the
test databases.
+ */
+ private void waitForTDengineReady() {
+ given().ignoreExceptions()
+ .await()
+ .pollInterval(1, TimeUnit.SECONDS)
+ .atMost(120, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(connection1.isValid(100));
+ Assertions.assertTrue(connection2.isValid(100));
+ verifyDatabaseCanBeCreated(connection1,
"seatunnel_ready_src");
+ verifyDatabaseCanBeCreated(connection2,
"seatunnel_ready_sink");
+ });
+ }
+
+ /**
+ * TDengine can report a healthy socket before the storage node is ready
to create databases.
+ */
+ private void verifyDatabaseCanBeCreated(Connection connection, String
databaseName)
+ throws SQLException {
+ try (Statement statement = connection.createStatement()) {
+ statement.execute("DROP DATABASE IF EXISTS " + databaseName);
+ statement.execute("CREATE DATABASE " + databaseName +
SINGLE_VGROUP_DATABASE_OPTIONS);
+ statement.execute("DROP DATABASE " + databaseName);
Review Comment:
**P3 — Unconditional `DROP DATABASE` will throw on second iteration**
```java
statement.execute("DROP DATABASE " + databaseName); // line 233
```
On the first iteration this works (the database was just created), but on
subsequent `verifyDatabaseCanBeCreated` calls the database may already be gone
(or never created if the previous `CREATE DATABASE` itself failed partway),
causing an exception. The Awaitility `ignoreExceptions()` swallows it, but the
logs become noisy and vgroup accounting may drift.
Suggest: `statement.execute("DROP DATABASE IF EXISTS " + databaseName);`
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-tdengine-e2e/src/test/java/org/apache/seatunnel/e2e/connector/tdengine/TDengineIT.java:
##########
@@ -224,6 +203,37 @@ private Connection createConnect(GenericContainer<?>
tdengineServer) {
return conn;
}
+ /**
+ * Waits until TDengine accepts JDBC traffic and can allocate vgroups for
new databases.
+ *
+ * <p>Port readiness alone does not prove that TDengine can create the
test databases.
+ */
+ private void waitForTDengineReady() {
Review Comment:
**P2 — `waitForTDengineReady` repeatedly creates/drops a database**
Each retry of `untilAsserted` runs `verifyDatabaseCanBeCreated` twice (once
per connection), which means up to two `CREATE DATABASE` + `DROP DATABASE` per
second of waiting. On the small CI runners this is the whole point of the test,
but it could be cheaper to verify readiness once (e.g. a single probe database
created in `startUp`, never dropped) and re-use the result.
Not blocking, but worth considering if startup flakiness persists.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java:
##########
@@ -1948,9 +1951,36 @@ public void
testKafkaToKafkaExactlyOnceOnStreaming(TestContainer container) {
}
return null;
});
+
+ // Use a committed sink record as the readiness signal. The SeaTunnel
Kafka source uses
+ // explicit partition assignment, so consumer-group membership is not
observable.
+ given().pollInterval(5, SECONDS)
+ .await()
+ .atMost(5, MINUTES)
+ .untilAsserted(
+ () -> {
+ ProducerRecord<byte[], byte[]> readinessRecord =
Review Comment:
**P2 — Readiness probe is not idempotent**
```java
ProducerRecord<byte[], byte[]> readinessRecord = new
ProducerRecord<>(producerTopic, null, readinessData.getBytes(...));
producer.send(readinessRecord);
```
Every retry inside `given().untilAsserted(...)` writes a new readiness
record. In the worst case (5min timeout / 5s poll) this is up to 60 records on
the producer topic and 60 corresponding committed records on the sink topic.
Concerns:
1. The probe data pollutes the source/topic and shifts transaction
boundaries for the EXACTLY_ONCE check.
2. The downstream `checkData` already adds `readinessData` to its ignore
list (line 2003) — that's a hint the probe is leaking into the assertion data.
Suggested fix: write the readiness record exactly once before starting the
job, and just poll the sink topic for it. Or use a unique key (so the broker
de-duplicates) plus a fixed value.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaIT.java:
##########
@@ -2590,23 +2624,35 @@ private List<String> getKafkaConsumerListData(String
topicName, long startOffset
long visibleEndOffsetExclusive =
consumer.endOffsets(Collections.singletonList(topicPartition))
.get(topicPartition);
+ long nextOffset = startOffset;
Long lastProcessedOffset = startOffset - 1;
int consecutiveEmptyPolls = 0;
Review Comment:
**P3 — `consecutiveEmptyPolls` and `nextOffset` interaction**
With the new READ_COMMITTED-aware loop:
```java
if (records.isEmpty()
&& currentPosition == nextOffset
&& consecutiveEmptyPolls >= 20) {
break;
}
nextOffset = currentPosition;
```
When records are empty but `currentPosition` keeps advancing (broker is
moving the LSO over aborted/control records but no visible records are
returned), `consecutiveEmptyPolls` is incremented while `nextOffset` keeps
being updated. The break condition `currentPosition == nextOffset` will not
trigger, so the loop continues until `nextOffset >= visibleEndOffsetExclusive`.
That looks correct for the exactly-once scenario, but please add a unit test
that exercises the "aborted transactions between startOffset and endOffset"
case so the loop's correctness is locked in.
##########
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-kafka-e2e/src/test/java/org/apache/seatunnel/e2e/connector/kafka/KafkaKerberosIT.java:
##########
@@ -479,5 +479,10 @@ public void tearDown() throws Exception {
if (kerberosContainer != null) {
kerberosContainer.close();
}
+ // KafkaKerberosIT sets global JVM JAAS/Krb5 properties during
startup. Clear them after
+ // the suite so the plain Kafka E2E tests in the same Maven JVM do not
inherit Kerberos
+ // authentication settings by mistake.
+ System.clearProperty("java.security.krb5.conf");
Review Comment:
**P2 — `tearDown` clearProperty may break re-entry**
```java
System.clearProperty("java.security.krb5.conf");
System.clearProperty("java.security.auth.login.config");
```
`startUp` calls `System.setProperty` only on the first `@BeforeAll`
invocation. If the JUnit lifecycle ever re-enters `startUp` after this
`tearDown` (e.g. some test harnesses re-run `@BeforeAll`/`@AfterAll` per fork,
or surefire forks re-execute the suite), the next run will start without those
properties set and Kerberos auth will fail.
Suggested fix: either always `System.setProperty(...)` in `startUp`
regardless of prior value, or stash the previous value in `startUp` and restore
it in `tearDown`.
--
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]
