[Bug]: HBase read could stuck at HBaseReader.advance indefinitely

[Abacn]

What happened?

There is report that HBaseIO read get stuck intermittently, possible due to failed connection never recovered

At here:

https://github.com/apache/beam/blob/16e7af2ea4148c65e3e1045e6c05f0d40f498db2/sdks/java/io/hbase/src/main/java/org/apache/beam/sdk/io/hbase/HBaseIO.java#L535C21-L535C21

Operation ongoing in bundle process_bundle-8636262876883182297-1613 for PTransform{id=Read-from-HBase-Read-HBaseSource--ParDo-BoundedSourceAsSDFWrapper--ParMultiDo-BoundedSourceAsSDFWrap/ProcessElementAndRestrictionWithSizing-ptransform-66, name=Read-from-HBase-Read-HBaseSource--ParDo-BoundedSourceAsSDFWrapper--ParMultiDo-BoundedSourceAsSDFWrap/ProcessElementAndRestrictionWithSizing-ptransform-66, state=process} for at
 least 01h15m05s without outputting or completing: at
 java.lang.Object.wait(Native Method) at
 java.lang.Object.wait(Object.java:460) at
 java.util.concurrent.TimeUnit.timedWait(TimeUnit.java:348) at
 org.apache.hadoop.hbase.client.ResultBoundedCompletionService.poll(ResultBoundedCompletionService.java:159) at
 org.apache.hadoop.hbase.client.ScannerCallableWithReplicas.call(ScannerCallableWithReplicas.java:198) at
 org.apache.hadoop.hbase.client.ScannerCallableWithReplicas.call(ScannerCallableWithReplicas.java:60) at
 org.apache.hadoop.hbase.client.RpcRetryingCaller.callWithoutRetries(RpcRetryingCaller.java:200) at
 org.apache.hadoop.hbase.client.ClientScanner.call(ClientScanner.java:320) at
 org.apache.hadoop.hbase.client.ClientScanner.loadCache(ClientScanner.java:403) at
 org.apache.hadoop.hbase.client.ClientScanner.next(ClientScanner.java:364) at
 org.apache.hadoop.hbase.client.AbstractClientScanner$1.hasNext(AbstractClientScanner.java:94) at
 org.apache.beam.sdk.io.hbase.HBaseIO$HBaseReader.advance(HBaseIO.java:514) at
 org.apache.beam.sdk.io.Read$BoundedSourceAsSDFWrapperFn$BoundedSourceAsSDFRestrictionTracker.tryClaim(Read.java:366) at
 ...
 java.lang.Thread.run(Thread.java:750)

Shall we add a timeout to advance?

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[github-actions]

This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.

[mxtymoshyk]

Still open, and I went digging into it. I think the diagnosis is a little different from "failed connection never recovered" -- the connection may well be healthy and the scan may well be making progress. The problem is that Beam has no way to tell.

advance() (HBaseIO.java:562) blocks inside iter.hasNext(), which lands in HBase's ClientScanner.loadCache(). Reading that method in 2.6.5, the version sdks/java/io/hbase/build.gradle pins today:

The individual RPC is bounded. loadCache issues each request through call(callable, caller, scannerTimeout, true) (ClientScanner.java:461), which goes on to callWithoutRetries with hbase.client.scanner.timeout.period, 60s by default. Both polls inside ScannerCallableWithReplicas.call are bounded too, and the second one throws "Failed to get result within timeout" rather than waiting. So no single request can sit there forever.

What is unbounded is the loop around it. loadCache is a bare for (;;) at ClientScanner.java:453, and retriesLeft only decrements on the DoNotRetryIOException path at :474. The way through it without touching any counter is the heartbeat path at :515. When the region server is still scanning but has not matched a row yet, it returns a heartbeat message; if the client-side cache is empty, the loop falls through to the bottom and goes around again. countdown is untouched because no rows arrived, remainingResultSize is untouched for the same reason, and retriesLeft never came into play because nothing threw. A scan with a selective filter over a large region can stay in there for as long as the server keeps looking, with every RPC completing successfully inside its timeout the whole time. The stack in the description is a single sample, so landing in the poll is consistent with going around that loop repeatedly rather than with one call hanging.

The awkward consequence is that no HBase client setting bounds this. HBaseIO hands the user's Configuration straight to ConnectionFactory.createConnection (HBaseIO.java:546) and the module sets no timeouts of its own anywhere, so hbase.client.scanner.timeout.period and hbase.client.retries.number are already reachable by users. They just do not apply to this case. Worth knowing before anyone offers them as the workaround.

From HBase's side nothing is going wrong here. Heartbeats exist precisely so that a slow server-side scan does not trip the scanner timeout. The problem sits on Beam's side of the boundary: while hasNext() is in that loop the reader cannot report progress, cannot be split, and cannot be cancelled, which is what the "operation ongoing ... without outputting or completing" message is really describing.

On the fix itself, a timeout in advance() is possible but more constrained than it first looks. advance() returning false means end of input, so a timeout has no way to say "still working" -- it would have to throw and fail the bundle. That is a real behaviour change, since a genuinely slow but healthy scan would start failing, and it needs a thread per reader to interrupt. HBase RPC threads do not reliably honour interrupts, so cancelled work can be left running behind it.

There is a second option I would put ahead of that, because HBase already has an answer for this exact situation. Scan.setNeedCursorResult(true) makes the server return a cursor Result on heartbeat instead of nothing, and the client then breaks out of the loop at ClientScanner.java:528-531 instead of spinning. advance() would need to recognise Result.isCursor(), skip it as a non-row, and use getCursor().getRow() to move the range tracker forward. Beam gets control back on every heartbeat, which restores progress reporting and cancellation without changing when a read is allowed to fail. I checked and all three of those APIs are present in the shaded client we pin today. The catch is that it needs an HBase 2.0+ server, so it would have to be opt-in and would quietly do nothing against an older one.

Those two are not the same fix and I do not want to guess which you want: one bounds how long a read may take, the other bounds how long Beam is blind for. They compose fine if you want the cursor for control and a timeout as a backstop, but that is a bigger change than either alone.

Testing looks cheap whichever way it goes. HBaseIOCloseTest already mocks ResultScanner and Connection, so a blocking mock iterator would give a deterministic test without needing the mini cluster that HBaseIOTest spins up.

Happy to write whichever you prefer. Not claiming the issue in the meantime.