rkhachatryan commented on a change in pull request #12292:
URL: https://github.com/apache/flink/pull/12292#discussion_r431115232
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java
##########
@@ -180,17 +177,33 @@ private void doComplete(boolean precondition,
RunnableWithException complete, Ru
}
private <I, H extends AbstractChannelStateHandle<I>> void complete(
+ StreamStateHandle underlying,
CompletableFuture<Collection<H>> future,
Map<I, List<Long>> offsets,
- BiFunction<I, List<Long>, H> buildHandle) {
+ TriFunction<I, StreamStateHandle, List<Long>, H>
buildHandle) throws IOException {
final Collection<H> handles = new ArrayList<>();
for (Map.Entry<I, List<Long>> e : offsets.entrySet()) {
- handles.add(buildHandle.apply(e.getKey(),
e.getValue()));
+ handles.add(createHandle(underlying, buildHandle, e));
}
future.complete(handles);
LOG.debug("channel state write completed, checkpointId: {},
handles: {}", checkpointId, handles);
}
+ private <I, H extends AbstractChannelStateHandle<I>> H createHandle(
+ StreamStateHandle underlying,
+ TriFunction<I, StreamStateHandle, List<Long>, H>
buildHandle,
+ Map.Entry<I, List<Long>> e) throws IOException {
+ if (underlying instanceof ByteStreamStateHandle) {
+ ByteStreamStateHandle byteHandle =
(ByteStreamStateHandle) underlying;
+ return buildHandle.apply(
+ e.getKey(),
+ new
ByteStreamStateHandle(randomUUID().toString(),
serializer.extractAndMerge(byteHandle.getData(), e.getValue())),
+ singletonList(serializer.getHeaderLength()));
+ } else {
Review comment:
Sure. Created a JIRA ticket:
https://issues.apache.org/jira/browse/FLINK-17972
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
