[FLINK-29971] Hbase sink will lose data at extreme case

• Type: Bug
• Status: Open
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 1.13.6, 1.15.3
• Component/s: Connectors / HBase
• Labels: pull-request-available stale-assigned
• Assignee: wenchao.wu
• Reporter: wenchao.wu
• Created: 10/Nov/22 08:24
• Updated: 14/Aug/23 22:35

Description

Situation:

When I use kafka as source and hbase as sink but the hbase table I didn't have the permission, I send data to kafka one message with a long time gap.

In this situation the normal result will be when trigger checkpoint the job will failed. But actually the jobs will continue to run and can trigger checkpoint successfully.

Analysis

The hbase sink will throw exception in checkErrorAndRethrow() funciton. And this function will be called in two function, invoke() and flush(). Beside invoke(), flush() will be called at two place, one is snapshot(), one is in the scheduledThread as the follow snippet of code:

We can see that in the  scheduledThread the exception throw by flush() will be catch and reset to failureThrowable.

So if there's no message come, the only way to throw the exception is in snapshot(). But the snapshot function call flush() is conditional as  the follow snippet of code:

But the scheduledThread will called flush() periodically and set numPendingRequests as 0.

So if no other message comes the snapshot will run successfully which means the checkpoint will be success but that message was not written to hbase, the message is loss.

 

Solution

I think the reason is that when trigger checkpoint and call snapshot function, need to call checkErrorAndRethrow() first as follow: 

Issue Links

• links to: GitHub Pull Request #21334