[INLONG-5169][Sort] Fix race condition issue of HBaseSinkFunction metric collection

[ifndef-SleePy]

Prepare a Pull Request

• Fixes [Bug][Sort] Metrics of HBaseSinkFunction are not collected accurately #5169

Motivation

• Fix race condition issue of HBaseSinkFunction metric collection

Modifications

Describe the modifications you've done.

Verifying this change

(Please pick either of the following options)

• This change is a trivial rework/code cleanup without any test coverage.

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not applicable)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation

[dockerzhang]

@yunqingmoswu @thexiay PTAL, thanks.

[gong]

Maybe we can define a ConcurrentCounter

[yunqingmoswu]

@yunqingmoswu @thexiay PTAL, thanks.

done

[ifndef-SleePy]

Hi guys, thanks for reviewing. First, I'm sorry that I didn't explain my points behind the PR.

As you can see, all the comments are focused on the scenario of flushing failure. I think we don't need care about the metrics after flushing failure. Because

1. If flushing fails, the job would falls into failover scenario.
2. Metric accuracy can not be guaranteed when failover happens. The consistency mode of HBaseSinkFunction is at-least-once. So some of produced records would probably be re-produced again. However the metric collection would not be reverted. All of these records would be counted.
3. Moreover, nobody knows how many records have been written into HBase Region Server when flushing fails. Flushing of BufferedMutator is not a transactional operation. If I remember correctly, HBase client spawns RPC calls to several Region Server concurrently. Any failure RPC call leads to a global failure. So when failure happens, we don't know how many RPC succeeded. It might be none, some or all records have been flushed successfully.

My points here are:

1. We have to guarantee the metric accuracy if no failure happens. That's what we do in this PR.
2. Metric is not accurate if flushing fails, no matter what we do. So we don't need to involve any synchronization of these counters. Just make it as cheaper as possible.
3. A counter records the number of flushing failure times is meaningful. But I'm a bit confused of the counter name "dirty", so I removed the these ambiguous counters. Maybe we could introduce another counter to record the failure times.

What do you guys think of this? @EMsnap @gong @yunqingmoswu

[yunqingmoswu]

Hi guys, thanks for reviewing. First, I'm sorry that I didn't explain my points behind the PR.

As you can see, all the comments are focused on the scenario of flushing failure. I think we don't need care about the metrics after flushing failure. Because

1. If flushing fails, the job would falls into failover scenario.
2. Metric accuracy can not be guaranteed when failover happens. The consistency mode of HBaseSinkFunction is at-least-once. So some of produced records would probably be re-produced again. However the metric collection would not be reverted. All of these records would be counted.
3. Moreover, nobody knows how many records have been written into HBase Region Server when flushing fails. Flushing of BufferedMutator is not a transactional operation. If I remember correctly, HBase client spawns RPC calls to several Region Server concurrently. Any failure RPC call leads to a global failure. So when failure happens, we don't know how many RPC succeeded. It might be none, some or all records have been flushed successfully.

My points here are:

1. We have to guarantee the metric accuracy if no failure happens. That's what we do in this PR.
2. Metric is not accurate if flushing fails, no matter what we do. So we don't need to involve any synchronization of these counters. Just make it as cheaper as possible.
3. A counter records the number of flushing failure times is meaningful. But I'm a bit confused of the counter name "dirty", so I removed the these ambiguous counters. Maybe we could introduce another counter to record the failure times.

What do you guys think of this? @EMsnap @gong @yunqingmoswu

As far as I know, the underlying logic of refresh should generate hfile first and then load, then refresh will only have this batch of data either all visible or all lost. If I understand correctly, then after refreshing the statistics, the data obtained will be more accurate, not only for dirty data, but also for normal sync data.

[gong]

@ifndef-SleePy hello, here define of dirty data is write fail, and we need data records and data size.

[ifndef-SleePy]

Hi guys, thanks for reviewing. First, I'm sorry that I didn't explain my points behind the PR.
As you can see, all the comments are focused on the scenario of flushing failure. I think we don't need care about the metrics after flushing failure. Because

1. If flushing fails, the job would falls into failover scenario.
2. Metric accuracy can not be guaranteed when failover happens. The consistency mode of HBaseSinkFunction is at-least-once. So some of produced records would probably be re-produced again. However the metric collection would not be reverted. All of these records would be counted.
3. Moreover, nobody knows how many records have been written into HBase Region Server when flushing fails. Flushing of BufferedMutator is not a transactional operation. If I remember correctly, HBase client spawns RPC calls to several Region Server concurrently. Any failure RPC call leads to a global failure. So when failure happens, we don't know how many RPC succeeded. It might be none, some or all records have been flushed successfully.

My points here are:

1. We have to guarantee the metric accuracy if no failure happens. That's what we do in this PR.
2. Metric is not accurate if flushing fails, no matter what we do. So we don't need to involve any synchronization of these counters. Just make it as cheaper as possible.
3. A counter records the number of flushing failure times is meaningful. But I'm a bit confused of the counter name "dirty", so I removed the these ambiguous counters. Maybe we could introduce another counter to record the failure times.

What do you guys think of this? @EMsnap @gong @yunqingmoswu

As far as I know, the underlying logic of refresh should generate hfile first and then load, then refresh will only have this batch of data either all visible or all lost. If I understand correctly, then after refreshing the statistics, the data obtained will be more accurate, not only for dirty data, but also for normal sync data.

1. The HFile way you described is called "bulk loading" [1]. That's not the way worked here. BufferedMutator would not trigger bulk loading, there is another API for bulk loading. It would go through WAL and MemStore to be ingested into Region Server.
2. Bulk loading also could not guarantee "this batch of data either all visible or all lost", because there might be several HFile involved in the batch operation.
3. HBase could guarantee consistency in a row, but not cross rows, please check the ACID docs [2], "APIs that mutate several rows will not be atomic across the multiple rows.".

[1] https://hbase.apache.org/book.html#arch.bulk.load
[2] https://hbase.apache.org/acid-semantics.html

[yunqingmoswu]

It is possible to report indicators before and after flushing, but it is recommended to report indicators after flushing.

[yunqingmoswu]

Hi guys, thanks for reviewing. First, I'm sorry that I didn't explain my points behind the PR.
As you can see, all the comments are focused on the scenario of flushing failure. I think we don't need care about the metrics after flushing failure. Because

1. If flushing fails, the job would falls into failover scenario.
2. Metric accuracy can not be guaranteed when failover happens. The consistency mode of HBaseSinkFunction is at-least-once. So some of produced records would probably be re-produced again. However the metric collection would not be reverted. All of these records would be counted.
3. Moreover, nobody knows how many records have been written into HBase Region Server when flushing fails. Flushing of BufferedMutator is not a transactional operation. If I remember correctly, HBase client spawns RPC calls to several Region Server concurrently. Any failure RPC call leads to a global failure. So when failure happens, we don't know how many RPC succeeded. It might be none, some or all records have been flushed successfully.

My points here are:

1. We have to guarantee the metric accuracy if no failure happens. That's what we do in this PR.
2. Metric is not accurate if flushing fails, no matter what we do. So we don't need to involve any synchronization of these counters. Just make it as cheaper as possible.
3. A counter records the number of flushing failure times is meaningful. But I'm a bit confused of the counter name "dirty", so I removed the these ambiguous counters. Maybe we could introduce another counter to record the failure times.

What do you guys think of this? @EMsnap @gong @yunqingmoswu

As far as I know, the underlying logic of refresh should generate hfile first and then load, then refresh will only have this batch of data either all visible or all lost. If I understand correctly, then after refreshing the statistics, the data obtained will be more accurate, not only for dirty data, but also for normal sync data.

1. The HFile way you described is called "bulk loading" [1]. That's not the way worked here. BufferedMutator would not trigger bulk loading, there is another API for bulk loading. It would go through WAL and MemStore to be ingested into Region Server.
2. Bulk loading also could not guarantee "this batch of data either all visible or all lost", because there might be several HFile involved in the batch operation.
3. HBase could guarantee consistency in a row, but not cross rows, please check the ACID docs [2], "APIs that mutate several rows will not be atomic across the multiple rows.".

[1] https://hbase.apache.org/book.html#arch.bulk.load [2] https://hbase.apache.org/acid-semantics.html

Thanks for clarifying.

[ifndef-SleePy]

After an offline communication with @yunqingmoswu , we reached an agreement that just keep the origin implementation with thread-safe counter. Because it's bases on production requirement, not technical.

1. We have to keep the meaningless "dirty" prefix counter till someday the production manager realizes that it's meaningless.
2. Summing the counter after flushing is better than before flushing although neither of them could guarantee accuracy. Because we don't want to meet the scenario that user asks a question why there exists successful counter in metric system however there is no data in HBase.

[gong]

please resolve conflic@ifndef-SleePy

[ifndef-SleePy]

I've re-created the PR after resolving conflicts.

[yunqingmoswu]

LGTM