[RFC][CELEBORN-894] Add support for end to end Integrity Checks #3062
Conversation
gauravkm
changed the title
|
Glad to see your PR. If it's ready you can remove the draft label. |
gauravkm
changed the title
|
Hi @FMX |
Code review will be done within this week. |
| @@ -377,7 +377,8 @@ private void close() throws IOException, InterruptedException { | |||
| updateRecordsWrittenMetrics(); | |||
|
|
|||
| long waitStartTime = System.nanoTime(); | |||
| shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers); | |||
| int bytesWritten = shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers, numMappers); | |||
| writeMetrics.incBytesWritten(bytesWritten); | |||
There was a problem hiding this comment.
Should not add the bytes written to spark metrics. Because the write metric has the correct value of written bytes.
| @@ -32,6 +33,7 @@ public class PushState { | |||
| private final int pushBufferMaxSize; | |||
| public AtomicReference<IOException> exception = new AtomicReference<>(); | |||
| private final InFlightRequestTracker inFlightRequestTracker; | |||
| private final ConcurrentHashMap<Integer, CommitMetadata> commitMetadataMap = new ConcurrentHashMap<>(); | |||
There was a problem hiding this comment.
This map is always empty. Seems that you forget to update this maps.
| int bytes = 0; | ||
|
|
||
| for (int partitionId = 0; partitionId < numPartitions; partitionId++) { | ||
| CommitMetadata metadata = metadataMap.getOrDefault(partitionId, new CommitMetadata()); |
There was a problem hiding this comment.
Here will always get empty commit meta data.
|
gentle ping @gauravkm |
|
@gaoyajun02 Yes. I am still working on this. We (at Stripe) realized that the implementation needs to be a lot more comprehensive and thorough for the checks to be meaningful and provide confidence. We are internally testing out the new implementation which I will then open for review in OSS as well. |
What changes were proposed in this pull request?
CELEBORN-894
End to End integrity checks reference implementation. Not looking to merge this yet. Looking for high level feedback on the approach
Why are the changes needed?
https://docs.google.com/document/d/1YqK0kua-5rMufJw57kEIrHHGbLnAF9iXM5GdDweMzzg/edit?tab=t.0
Does this PR introduce any user-facing change?
How was this patch tested?