pnowojski commented on a change in pull request #12292:
URL: https://github.com/apache/flink/pull/12292#discussion_r430645868
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java
##########
@@ -25,8 +25,10 @@
import org.apache.flink.runtime.state.InputChannelStateHandle;
Review comment:
Thanks, now it's better :) Could you copy this updated description to
commit message and jira ticket?
##########
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:
I think it's better now, but I still can not put a finger on a feeling
that something might not be entirely correct.
Why do we have multiple handles per channel, instead of a single handle that
has multiple channels?
----------------------------------------------------------------
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]
