[Bug] Sink tests race their sinks' async flush: fixed Thread.sleep before verify()

[david-streamlio]

Summary

An audit for timing flakes turned up 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. When CI is slow, the flush has not run and the assertion fails — the sink is fine; the test raced it.

This is exactly the shape already fixed twice today:

• InfluxDBSinkTest.testOpenWriteCloseJson (v2) — failed on unrelated PR [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 ([improve][test] Add integration tests for the InfluxDB sink connector #53).
• PulsarSchemaHistoryTest.setup — failed on unrelated PR [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 ([Bug] Flaky test: PulsarSchemaHistoryTest.setup fails with 'Tenant does not exist' #83).

Both took down PRs that never touched the code in question.

Findings

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

write5000Test is the sharpest: it asserts 5000 acks have landed within a hard-coded 2 seconds.

Lower priority / not flaky

• redis — RedisSinkTest sleeps 1000 ms but asserts nothing afterwards; close() does the flushing. Pointless, not racy.
• jdbc/sqlite — SqliteUtils sleeps 100 ms inside a retry loop. Benign.
• azure-data-explorer — two Thread.sleep(40000) calls, but the test is credential-gated and never runs in CI ([improve][test] Azure Data Explorer E2E test is always skipped in CI — add a runnable integration test #49).
• file — the 25 sleeps are mostly deliberate 30-second soak windows; the hand-off races were fixed in [fix][io] Fix duplicate file offers and flaky tests in file connector #41 and the drain loop in [fix][test] Bound record reads in Debezium MySQL and Postgres tests #71.

Fix

Replace each fixed sleep with an Awaitility poll on the same condition (awaitility is already on every module's test classpath via the shared java-conventions plugin). The assertions stay identical; only the waiting changes. This removes the race without weakening any test — and speeds them up, since the poll returns as soon as the condition holds rather than always sleeping the full duration.