Flink: Fix hash code comparison for requesting global statistics in DataStatisticsCoordinator #13827
Conversation
|
Hi @stevenzwu @pvary , could you please help review this PR and verify whether it is appropriate? Thank you very much! |
| @@ -277,7 +277,7 @@ private void handleRequestGlobalStatisticsEvent(int subtask, RequestGlobalStatis | |||
| if (globalStatistics != null) { | |||
| runInCoordinatorThread( | |||
| () -> { | |||
| if (event.signature() != null && event.signature() != globalStatistics.hashCode()) { | |||
| if (event.signature() != null && event.signature() == globalStatistics.hashCode()) { | |||
There was a problem hiding this comment.
thx for catching the bug. I think your fix is correct.
Should we also fix the log message as following?
Skip responding to statistics request from subtask {}, as the operator task already holds the same global statistics
There was a problem hiding this comment.
Yes, I have change it now.
| dataStatisticsCoordinator.handleEventFromOperator( | ||
| 0, 0, new RequestGlobalStatisticsEvent(correctSignature)); | ||
|
|
||
| Thread.sleep(200); |
There was a problem hiding this comment.
I know we are waiting for 200 ms to confirm no response is sent in this case. We can probably replace the sleep with waitForCoordinatorToProcessActions. Then we can immediately assert the sent events count hasn't changed.
There was a problem hiding this comment.
Thanks for point is out, I have change it .
| .atMost(Duration.ofSeconds(10)) | ||
| .until(() -> receivingTasks.getSentEventsForSubtask(0).size() == 2); | ||
|
|
||
| // signature is right |
There was a problem hiding this comment.
nit: maybe the comment can be changed to following?
Simulate the scenario where a subtask send global statistics request with the same hash code. The coordinator would skip the response after comparing the request contained hash code with latest global statistics hash code.
stevenzwu
changed the title
|
thanks @Guosmilesmile for catching and fixing the bug |
In
DataStatisticsCoordinator, when handling theRequestGlobalStatisticsEvent, the coordinator should skip responding to the subtask if the event's signature matches thehashCodeof the currentglobalStatistics.The current implementation is incorrect—this PR fixes that behavior.