[#808] improvement(spark): Verify the number of written records to enhance data correctness

[rickyma]

What changes were proposed in this pull request?

Verify the number of written records to enhance data accuracy.
Make sure all data records are sent by clients.
Make sure bugs like #714 will never be introduced into the code.

Why are the changes needed?

A follow-up PR for #848.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing UTs.

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 85.71429% with 1 lines in your changes are missing coverage. Please review.

Project coverage is 54.57%. Comparing base (d8aedf3) to head (1c4f86b).

Files	Patch %	Lines
...pache/spark/shuffle/writer/WriteBufferManager.java	75.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #1558      +/-   ##
============================================
+ Coverage     53.65%   54.57%   +0.91%     
- Complexity     2818     2819       +1     
============================================
  Files           436      416      -20     
  Lines         24549    22195    -2354     
  Branches       2080     2080              
============================================
- Hits          13172    12113    -1059     
+ Misses        10554     9330    -1224     
+ Partials        823      752      -71     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[github-actions]

Test Results

2 312 files  ±0  2 312 suites  ±0   4h 36m 26s ⏱️ +50s
  823 tests ±0    822 ✅ ±0   1 💤 ±0  0 ❌ ±0 
9 697 runs  ±0  9 683 ✅ ±0  14 💤 ±0  0 ❌ ±0 

Results for commit 1c4f86b. ± Comparison against base commit d8aedf3.

♻️ This comment has been updated with latest results.

[rickyma]

PTAL @jerqi @zuston

[zuston]

LGTM overall. Left some minor comments.

And I hope this could be covered by test cases. Like throwing exception when encountering records is not same.

[rickyma]

And I hope this could be covered by test cases. Like throwing exception when encountering records is not same.

I think the existing UTs have already covered this. Because this is a base method which will be used in many places.
If anything bad is introduced into the code in the future, I think the UTs won't pass.

[jerqi]

Maybe record count isn't enough. We should check block count, too.

[rickyma]

Maybe record count isn't enough. We should check block count, too.

BlockIds will be checked in the method checkBlockSendResult. If any block is failed or lost, we will know.

[zuston]

And I hope this could be covered by test cases. Like throwing exception when encountering records is not same.

I think the existing UTs have already covered this. Because this is a base method which will be used in many places. If anything bad is introduced into the code in the future, I think the UTs won't pass.

No I hope you could mock throwing exception when record is not same.

[rickyma]

And I hope this could be covered by test cases. Like throwing exception when encountering records is not same.

I think the existing UTs have already covered this. Because this is a base method which will be used in many places. If anything bad is introduced into the code in the future, I think the UTs won't pass.

No I hope you could mock throwing exception when record is not same.

I think it's not easy to mock this and might not be very meaningful. The issue in the past was due to a critical bug in the code itself, caused by concurrency problems, which led to data loss even after calling the method WriteBuffer.addRecord. If we revert the code to the previous problematic version and use the content of my current PR, the unit test would definitely fail.

The fact that the unit test with the current check passes actually indicates that the current code is working correctly.

[zuston]

Merged. Thanks @rickyma