[fix][test] Replace fixed sleeps with Awaitility polls in sink tests

[david-streamlio]

Fixes #85

Motivation

An audit for timing flakes found the same defect in four more modules. Each writes to a sink that flushes asynchronously, waits a fixed Thread.sleep, then asserts the flush happened. On a slow runner the flush has not run and the assertion fails — the sink is fine, the test raced it.

This shape has already failed two unrelated PRs today:

• InfluxDBSinkTest.testOpenWriteCloseJson (v2) broke [fix][io] jdbc: drain flush() iteratively to avoid StackOverflowError under sustained backlog #39 → fixed in [improve][test] Add InfluxDB v1 and v2 sink integration tests, deflake the v2 mock test #73
• PulsarSchemaHistoryTest.setup broke [fix][io] Fix DynamoDB source client construction with a custom endpoint #79 → fixed in [fix][test] Wait for the public tenant before creating the test namespace #84

Findings

Module	Test	Sleep	Then asserts	Async mechanism
hdfs3	HdfsSequentialSinkTest.write5000Test	2000 ms	verify(..., times(5000)).ack()	syncInterval background sync
hdfs3	HdfsSequentialSinkTest, HdfsTextSinkTest	2000 ms	verify(..., times(100)).ack()	same
mongo	MongoSinkTest × 5 sites	1000 ms	verify(..., times(n)).ack()/.fail()	flushExecutor.scheduleAtFixedRate
influxdb v1	InfluxDBGenericRecordSinkTest	1000 ms	verify(influxDB, times(1)).write(...)	BatchSink flush executor — the very class whose v2 sibling flaked
hbase	HbaseGenericRecordSinkTest	500 ms	reads the row back	HbaseAbstractSink batch flush

write5000Test is the sharpest: 5000 acks asserted within a hard-coded 2 seconds.

Modifications

Replace each fixed sleep with an Awaitility poll on the same condition. The assertions are unchanged — only the waiting is. awaitility is already on every module's test classpath via the shared java-conventions plugin. The tests also get faster, since a poll returns as soon as the condition holds rather than always sleeping the full duration.

Verifying this change

All four suites pass: ./gradlew :mongo:test :influxdb:test :hdfs3:test :hbase:test → BUILD SUCCESSFUL.

More importantly, the race is demonstrated, not asserted. Taking master's MongoSinkTest and shortening its sleep from 1000 ms to 1 ms — simulating a runner slow enough that the flush has not fired — reproduces the CI signature exactly:

MongoSinkTest > testWriteBadMessage FAILED
    Wanted but not invoked: ...
MongoSinkTest > testWriteGoodMessage FAILED
    Wanted but not invoked: ...

The Awaitility version passes with no sleep at all, because it waits on the condition rather than on a guessed duration.

Notes

• HbaseGenericRecordSinkTest is @Test(enabled = false) and does not run today, so its change is compile-only. Included so the pattern is not reintroduced when it is re-enabled.
• Deliberately not changed, with reasons, in [Bug] Sink tests race their sinks' async flush: fixed Thread.sleep before verify() #85: redis (sleeps but asserts nothing afterwards), jdbc/sqlite (sleep inside a retry loop), azure-data-explorer (credential-gated, never runs in CI — see [improve][test] Azure Data Explorer E2E test is always skipped in CI — add a runnable integration test #49), and file (its sleeps are deliberate soak windows; its races were fixed in [fix][io] Fix duplicate file offers and flaky tests in file connector #41 and [fix][test] Bound record reads in Debezium MySQL and Postgres tests #71).

[Copilot]

Pull request overview

This PR deflakes multiple sink unit tests by replacing fixed Thread.sleep(...) delays with Awaitility-based polling, ensuring assertions wait for asynchronous flush/ack behavior rather than racing it.

Changes:

• Replace fixed sleeps with Awaitility.await().atMost(...).untilAsserted(...) in MongoDB, InfluxDB v1, and HDFS3 sink tests.
• Update the HBase sink test (currently disabled) to poll for the row to appear instead of sleeping before reading it back.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java	Replace multiple Thread.sleep(1000) calls with Awaitility polling for ack()/fail() verifications.
influxdb/src/test/java/org/apache/pulsar/io/influxdb/v1/InfluxDBGenericRecordSinkTest.java	Replace Thread.sleep(1000) with Awaitility polling for async batch flush verification.
hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsTextSinkTest.java	Replace Thread.sleep(2000) with Awaitility polling for 100/5000 ack assertions.
hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/seq/HdfsSequentialSinkTest.java	Replace Thread.sleep(2000) with Awaitility polling for 100/5000 ack assertions.
hbase/src/test/java/org/apache/pulsar/io/hbase/sink/HbaseGenericRecordSinkTest.java	Replace Thread.sleep(500) with Awaitility polling until the written row is observable.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.