[Feature][Flink]add read checkpoint

dinky-admin/src/main/java/org/dinky/controller/FlinkConfController.java → dinky-admin/src/main/java/org/dinky/controller/FlinkController.java

@@ -19,13 +19,16 @@
 
 package org.dinky.controller;
 
+import org.dinky.data.model.CheckPointReadTable;
 import org.dinky.data.result.Result;
 import org.dinky.data.vo.CascaderVO;
+import org.dinky.flink.checkpoint.CheckpointRead;
 import org.dinky.utils.CascaderOptionsUtils;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -39,7 +42,15 @@
 @Slf4j
 @Api(tags = "Flink Conf Controller", hidden = true)
 @RequestMapping("/api/flinkConf")
-public class FlinkConfController {
+public class FlinkController {
+    protected static final CheckpointRead INSTANCE = new CheckpointRead();
+
+    @GetMapping("/readCheckPoint")
+    @ApiOperation("Read Checkpoint")
+    public Result<Map<String, Map<String, CheckPointReadTable>>> readCheckPoint(String path, String operatorId) {
+        return Result.data(INSTANCE.readCheckpoint(path, operatorId));
+    }
+
     @GetMapping("/configOptions")
     @ApiOperation("Query Flink Configuration Options")
     public Result<List<CascaderVO>> loadDataByGroup() {
@@ -71,6 +82,8 @@ public Result<List<CascaderVO>> loadDataByGroup() {
             "org.apache.flink.configuration.MetricOptions",
             "org.apache.flink.configuration.NettyShuffleEnvironmentOptions",
             "org.apache.flink.configuration.RestartStrategyOptions",
+            "org.apache.flink.yarn.configuration.YarnConfigOptions",
+            "org.apache.flink.kubernetes.configuration.KubernetesConfigOptions",
             "org.dinky.constant.CustomerConfigureOptions"
         };
         List<CascaderVO> dataList = new ArrayList<>();

dinky-admin/src/main/java/org/dinky/controller/StudioController.java

@@ -92,7 +92,7 @@ public Result<JobResult> executeSql(@RequestBody StudioExecuteDTO studioExecuteD
             jobResult.setSuccess(false);
             jobResult.setStatement(studioExecuteDTO.getStatement());
             jobResult.setError(ex.toString());
-            return Result.failed(jobResult, Status.EXECUTE_FAILED);
+            return Result.failed(jobResult, ex.toString());
         }
     }
 

dinky-admin/src/main/java/org/dinky/sse/SseEmitterUTF8.java

@@ -19,7 +19,6 @@
 
 package org.dinky.sse;
 
-import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 
 import org.springframework.http.HttpHeaders;
@@ -39,18 +38,16 @@ public SseEmitterUTF8(Long timeout) {
     @Override
     protected void extendResponse(ServerHttpResponse outputMessage) {
         super.extendResponse(outputMessage);
-
         HttpHeaders headers = outputMessage.getHeaders();
         headers.setContentType(new MediaType(MediaType.TEXT_EVENT_STREAM, StandardCharsets.UTF_8));
     }
 
     @Override
-    public void send(Object object, MediaType mediaType) throws IOException {
+    public synchronized void complete() {
         Boolean complete = (Boolean) ReflectUtil.getFieldValue(this, "complete");
         if (complete) {
-            log.warn("SseEmitter is complete, cannot send message: {}", object);
             return;
         }
-        super.send(object, mediaType);
+        super.complete();
     }
 }

dinky-client/dinky-client-base/src/main/java/org/dinky/flink/checkpoint/BaseCheckpointRead.java

@@ -0,0 +1,41 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.flink.checkpoint;
+
+import org.dinky.data.model.CheckPointReadTable;
+
+import org.apache.flink.runtime.state.ArrayListSerializer;
+import org.apache.flink.runtime.state.PartitionableListState;
+
+import java.util.Optional;
+
+import cn.hutool.core.util.ReflectUtil;
+
+public abstract class BaseCheckpointRead {
+    protected BaseCheckpointRead() {}
+
+    public abstract boolean isSourceCkp(PartitionableListState<?> partitionableListState);
+
+    public abstract Optional<CheckPointReadTable> create(PartitionableListState<?> partitionableListState);
+
+    protected static ArrayListSerializer<?> getArrayListSerializer(PartitionableListState<?> partitionableListState) {
+        return (ArrayListSerializer<?>) ReflectUtil.getFieldValue(partitionableListState, "internalListCopySerializer");
+    }
+}

dinky-client/dinky-client-base/src/main/java/org/dinky/flink/checkpoint/CheckpointRead.java

@@ -0,0 +1,152 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.flink.checkpoint;
+
+import org.dinky.data.model.CheckPointReadTable;
+
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.runtime.checkpoint.OperatorState;
+import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.state.OperatorBackendSerializationProxy;
+import org.apache.flink.runtime.state.OperatorStateHandle;
+import org.apache.flink.runtime.state.PartitionableListState;
+import org.apache.flink.runtime.state.RegisteredOperatorStateBackendMetaInfo;
+import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
+import org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshot;
+import org.apache.flink.state.api.ExistingSavepoint;
+import org.apache.flink.state.api.Savepoint;
+import org.apache.flink.state.api.runtime.metadata.SavepointMetadata;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.ReflectUtil;
+
+public class CheckpointRead implements CheckpointReadInterface {
+    @Override
+    public Map<String, Map<String, CheckPointReadTable>> readCheckpoint(String path, String operatorId) {
+        ClassLoader restoreClassLoader = Thread.currentThread().getContextClassLoader();
+        Map<String, Map<String, CheckPointReadTable>> result = new LinkedHashMap<>();
+        try {
+            ExistingSavepoint savepoint =
+                    Savepoint.load(ExecutionEnvironment.getExecutionEnvironment(), path, new HashMapStateBackend());
+            List<OperatorState> operatorStateList =
+                    ((SavepointMetadata) ReflectUtil.getFieldValue(savepoint, "metadata")).getExistingOperators();
+            OperatorState existingOperator = operatorStateList.stream()
+                    .filter(operatorState -> operatorState
+                            .getOperatorID()
+                            .equals(OperatorID.fromJobVertexID(JobVertexID.fromHexString(operatorId))))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalArgumentException("The corresponding operator ID was not found"));
+            Map<Integer, OperatorSubtaskState> subtaskStates = existingOperator.getSubtaskStates();
+            if (CollUtil.isNotEmpty(subtaskStates)) {
+                subtaskStates.forEach((k, v) -> {
+                    StateObjectCollection<OperatorStateHandle> managedOperatorState = v.getManagedOperatorState();
+                    Map<String, CheckPointReadTable> read = readState(restoreClassLoader, managedOperatorState);
+                    result.put("managedOperatorState", read);
+                });
+            }
+            return result;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static Map<String, CheckPointReadTable> readState(
+            ClassLoader restoreClassLoader, StateObjectCollection<OperatorStateHandle> managedOperatorState) {
+        Map<String, CheckPointReadTable> map = new LinkedHashMap<>();
+        OperatorBackendSerializationProxy backendSerializationProxy =
+                new OperatorBackendSerializationProxy(restoreClassLoader);
+        boolean isRead = false;
+        for (OperatorStateHandle operatorStateHandle : managedOperatorState) {
+            try (FSDataInputStream in =
+                    operatorStateHandle.getDelegateStateHandle().openInputStream()) {
+                if (!isRead) {
+                    backendSerializationProxy.read(new DataInputViewStreamWrapper(in));
+                    isRead = true;
+                }
+
+                operatorStateHandle.getStateNameToPartitionOffsets().forEach((key, value) -> {
+                    try {
+                        List<StateMetaInfoSnapshot> restoredOperatorMetaInfoSnapshots =
+                                backendSerializationProxy.getOperatorStateMetaInfoSnapshots();
+                        for (StateMetaInfoSnapshot stateMetaInfoSnapshot : restoredOperatorMetaInfoSnapshots) {
+                            String name = stateMetaInfoSnapshot.getName();
+                            if (!name.equals(key)) {
+                                continue;
+                            }
+                            TypeSerializerSnapshot<?> valueSerializer = stateMetaInfoSnapshot
+                                    .getSerializerSnapshotsImmutable()
+                                    .get("VALUE_SERIALIZER");
+
+                            PartitionableListState<?> partitionableListState = ReflectUtil.newInstance(
+                                    PartitionableListState.class,
+                                    new RegisteredOperatorStateBackendMetaInfo<>(stateMetaInfoSnapshot));
+                            ;
+                            deserializeOperatorStateValues(partitionableListState, in, value);
+                            partitionableListState.get().forEach(System.out::println);
+                            // get checkpoint data
+                            CheckpointReadFactory.getTable(partitionableListState)
+                                    .ifPresent(tableVO -> map.put(key, tableVO));
+                            break;
+                        }
+
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                });
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        return map;
+    }
+
+    protected static <S> void deserializeOperatorStateValues(
+            PartitionableListState<S> stateListForName,
+            FSDataInputStream in,
+            OperatorStateHandle.StateMetaInfo metaInfo)
+            throws IOException {
+
+        if (null != metaInfo) {
+            long[] offsets = metaInfo.getOffsets();
+            if (null != offsets) {
+                DataInputView div = new DataInputViewStreamWrapper(in);
+                TypeSerializer<S> serializer =
+                        stateListForName.getStateMetaInfo().getPartitionStateSerializer();
+                for (long offset : offsets) {
+                    in.seek(offset);
+                    stateListForName.add(serializer.deserialize(div));
+                }
+            }
+        }
+    }
+}

dinky-client/dinky-client-base/src/main/java/org/dinky/flink/checkpoint/CheckpointReadFactory.java

@@ -0,0 +1,51 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.flink.checkpoint;
+
+import org.dinky.data.model.CheckPointReadTable;
+import org.dinky.flink.checkpoint.base.BaseTypeCheckpointRead;
+import org.dinky.flink.checkpoint.pojo.PojoTypeCheckpointRead;
+import org.dinky.flink.checkpoint.source.CheckpointSourceRead;
+
+import org.apache.flink.runtime.state.PartitionableListState;
+
+import java.util.List;
+import java.util.Optional;
+
+import cn.hutool.core.collection.CollUtil;
+
+public class CheckpointReadFactory {
+    public static final List<? extends BaseCheckpointRead> FACTORY_LIST = CollUtil.newArrayList(
+            new BaseTypeCheckpointRead(), new CheckpointSourceRead(), new PojoTypeCheckpointRead());
+
+    public static Optional<CheckPointReadTable> getTable(PartitionableListState<?> partitionableListState) {
+        Iterable<?> objects = partitionableListState.get();
+        boolean empty = CollUtil.isEmpty(objects);
+        if (empty) {
+            return Optional.empty();
+        }
+        for (BaseCheckpointRead baseCheckpointRead : FACTORY_LIST) {
+            if (baseCheckpointRead.isSourceCkp(partitionableListState)) {
+                return baseCheckpointRead.create(partitionableListState);
+            }
+        }
+        return Optional.empty();
+    }
+}

dinky-client/dinky-client-base/src/main/java/org/dinky/flink/checkpoint/CheckpointReadInterface.java

@@ -0,0 +1,36 @@
+/*
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ */
+
+package org.dinky.flink.checkpoint;
+
+import org.dinky.data.model.CheckPointReadTable;
+
+import java.util.Map;
+
+public interface CheckpointReadInterface {
+    /**
+     * 读取checkpoint
+     * @param path Checkpoint路径
+     * @param operatorId 执行id
+     * @return stateType -> (stateName -> CheckPointReadTable)
+     */
+    default Map<String, Map<String, CheckPointReadTable>> readCheckpoint(String path, String operatorId) {
+        throw new UnsupportedOperationException("readCheckpoint not implemented");
+    }
+}