This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 2ac8d846 [fix][test] Replace fixed sleeps with Awaitility polls in
sink tests (#86)
2ac8d846 is described below
commit 2ac8d8462bc1cc7a8bfd1e6a6f82f95ee8e113bb
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Fri Jul 10 08:50:40 2026 -0700
[fix][test] Replace fixed sleeps with Awaitility polls in sink tests (#86)
* [fix][test] Replace fixed sleeps with Awaitility polls in sink tests
Five sink tests wrote to a sink that flushes asynchronously, waited a
fixed Thread.sleep, then asserted the flush had happened. When CI is
slow the flush has not run and the assertion fails: the sink is fine,
the test raced it. This is the shape that failed unrelated PRs twice
today, in InfluxDBSinkTest (v2) and PulsarSchemaHistoryTest.
Poll the same conditions with Awaitility instead. The assertions are
unchanged; only the waiting is. The tests also get faster, since a poll
returns as soon as the condition holds.
- mongo: MongoSinkTest, 5 sites
- influxdb: InfluxDBGenericRecordSinkTest (v1) — sibling of the v2 test
already deflaked
- hdfs3: HdfsSequentialSinkTest, HdfsTextSinkTest — one asserted 5000
acks within a hard-coded 2 seconds
- hbase: HbaseGenericRecordSinkTest — currently @Test(enabled = false),
so this is a compile-only change there
* Potential fix for pull request finding
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
.../io/hbase/sink/HbaseGenericRecordSinkTest.java | 11 ++++---
.../io/hdfs3/sink/seq/HdfsSequentialSinkTest.java | 12 +++++---
.../pulsar/io/hdfs3/sink/seq/HdfsTextSinkTest.java | 12 +++++---
.../influxdb/v1/InfluxDBGenericRecordSinkTest.java | 8 +++--
.../apache/pulsar/io/mongodb/MongoSinkTest.java | 36 ++++++++++++----------
5 files changed, 48 insertions(+), 31 deletions(-)
diff --git
a/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
b/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
index e0cef8cf..99e5936f 100644
---
a/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
+++
b/hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java
@@ -24,6 +24,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.client.Get;
@@ -46,6 +47,7 @@ import org.apache.pulsar.functions.source.PulsarRecord;
import org.apache.pulsar.functions.source.PulsarSourceConfig;
import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.hbase.TableUtils;
+import org.awaitility.Awaitility;
import org.mockito.Mock;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -155,12 +157,13 @@ public class HbaseGenericRecordSinkTest {
// write should success.
sink.write(record);
log.info("executed write");
- // sleep to wait backend flush complete
- Thread.sleep(500);
-
- // value has been written to hbase table, read it out and verify.
+ // The sink flushes on a background executor; poll for the row rather
than racing it.
Table table = TableUtils.getTable(map);
Get scan = new Get(Bytes.toBytes(obj.getRowKey()));
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .until(() -> !table.get(scan).isEmpty());
+
Result result = table.get(scan);
byte[] byteName = result.getValue(Bytes.toBytes(familyName),
Bytes.toBytes(name));
byte[] byteAddress = result.getValue(Bytes.toBytes(familyName),
Bytes.toBytes(address));
diff --git
a/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsSequentialSinkTest.java
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsSequentialSinkTest.java
index 8876c1cd..f1e0b08e 100644
---
a/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsSequentialSinkTest.java
+++
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsSequentialSinkTest.java
@@ -19,9 +19,11 @@
package org.apache.pulsar.io.hdfs3.sink.seq;
import static org.mockito.Mockito.times;
+import java.util.concurrent.TimeUnit;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertNotNull;
import org.apache.pulsar.io.hdfs3.sink.AbstractHdfsSinkTest;
+import org.awaitility.Awaitility;
import org.testng.SkipException;
import org.testng.annotations.Test;
@@ -42,8 +44,9 @@ public class HdfsSequentialSinkTest extends
AbstractHdfsSinkTest<Long, String> {
assertNotNull(sink);
send(100);
- Thread.sleep(2000);
- verify(mockRecord, times(100)).ack();
+ // The sink syncs on a background interval; poll rather than racing it.
+ Awaitility.await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(100)).ack());
sink.close();
}
@@ -57,8 +60,9 @@ public class HdfsSequentialSinkTest extends
AbstractHdfsSinkTest<Long, String> {
assertNotNull(sink);
send(5000);
- Thread.sleep(2000);
- verify(mockRecord, times(5000)).ack();
+ // The sink syncs on a background interval; poll rather than racing it.
+ Awaitility.await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(5000)).ack());
sink.close();
}
diff --git
a/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsTextSinkTest.java
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsTextSinkTest.java
index eb8a9e3d..c7f07eae 100644
---
a/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsTextSinkTest.java
+++
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsTextSinkTest.java
@@ -21,7 +21,9 @@ package org.apache.pulsar.io.hdfs3.sink.seq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.testng.Assert.assertNotNull;
+import java.util.concurrent.TimeUnit;
import org.apache.pulsar.io.hdfs3.sink.AbstractHdfsSinkTest;
+import org.awaitility.Awaitility;
import org.testng.SkipException;
import org.testng.annotations.Test;
@@ -43,8 +45,9 @@ public class HdfsTextSinkTest extends
AbstractHdfsSinkTest<String, String> {
assertNotNull(mockRecord);
send(100);
- Thread.sleep(2000);
- verify(mockRecord, times(100)).ack();
+ // The sink syncs on a background interval; poll rather than racing it.
+ Awaitility.await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(100)).ack());
sink.close();
}
@@ -59,8 +62,9 @@ public class HdfsTextSinkTest extends
AbstractHdfsSinkTest<String, String> {
assertNotNull(mockRecord);
send(5000);
- Thread.sleep(2000);
- verify(mockRecord, times(5000)).ack();
+ // The sink syncs on a background interval; poll rather than racing it.
+ Awaitility.await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(5000)).ack());
sink.close();
}
diff --git
a/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v1/InfluxDBGenericRecordSinkTest.java
b/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v1/InfluxDBGenericRecordSinkTest.java
index 6b7af26a..849f8cb2 100644
---
a/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v1/InfluxDBGenericRecordSinkTest.java
+++
b/influxdb/src/test/java/org/apache/pulsar/io/influxdb/v1/InfluxDBGenericRecordSinkTest.java
@@ -26,6 +26,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.Maps;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Message;
@@ -39,6 +40,7 @@ import
org.apache.pulsar.client.impl.schema.generic.GenericSchemaImpl;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.source.PulsarRecord;
import org.apache.pulsar.io.core.SinkContext;
+import org.awaitility.Awaitility;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.mockito.Mock;
@@ -150,8 +152,8 @@ public class InfluxDBGenericRecordSinkTest {
influxSink.write(record);
- Thread.sleep(1000);
-
- verify(influxDB, times(1)).write(any(BatchPoints.class));
+ // BatchSink flushes on a background executor; poll rather than racing
it.
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(influxDB,
times(1)).write(any(BatchPoints.class)));
}
}
diff --git
a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
index ea62e546..0da91d40 100644
--- a/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
+++ b/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
@@ -42,8 +42,10 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.SinkContext;
+import org.awaitility.Awaitility;
import org.bson.BsonDocument;
import org.mockito.Mock;
import org.reactivestreams.Publisher;
@@ -149,9 +151,9 @@ public class MongoSinkTest {
sink.open(map, mockSinkContext);
sink.write(mockRecord);
- Thread.sleep(1000);
-
- verify(mockRecord, times(1)).fail();
+ // The sink flushes on a background executor; poll rather than racing
it.
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(1)).fail());
}
@Test
@@ -161,9 +163,9 @@ public class MongoSinkTest {
sink.open(map, mockSinkContext);
sink.write(mockRecord);
- Thread.sleep(1000);
-
- verify(mockRecord, times(1)).ack();
+ // The sink flushes on a background executor; poll rather than racing
it.
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(1)).ack());
}
@Test
@@ -175,10 +177,12 @@ public class MongoSinkTest {
sink.write(mockRecord);
sink.write(mockRecord);
- Thread.sleep(1000);
-
- verify(mockRecord, times(2)).ack();
- verify(mockRecord, times(1)).fail();
+ // The sink flushes on a background executor; poll rather than racing
it.
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ verify(mockRecord, times(2)).ack();
+ verify(mockRecord, times(1)).fail();
+ });
}
@Test
@@ -188,9 +192,9 @@ public class MongoSinkTest {
sink.open(map, mockSinkContext);
sink.write(mockRecord);
- Thread.sleep(1000);
-
- verify(mockRecord, times(1)).fail();
+ // The sink flushes on a background executor; poll rather than racing
it.
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(1)).fail());
}
@Test
@@ -200,8 +204,8 @@ public class MongoSinkTest {
sink.open(map, mockSinkContext);
sink.write(mockRecord);
- Thread.sleep(1000);
-
- verify(mockRecord, times(1)).fail();
+ // The sink flushes on a background executor; poll rather than racing
it.
+ Awaitility.await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(mockRecord, times(1)).fail());
}
}
