rkhachatryan commented on a change in pull request #14943: URL: https://github.com/apache/flink/pull/14943#discussion_r578414166
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/delegate/DelegatedStateBackend.java ########## @@ -0,0 +1,29 @@ +/* + * 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.apache.flink.runtime.state.delegate; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.state.StateBackend; + +@Internal +/** + * An interface for DelegatedStateBackend. A state backend to be delegated must implement this + * interface. + */ +public interface DelegatedStateBackend extends StateBackend {} Review comment: I'm curious why didn't you add `unwrap` method to the `StateBackend` interface? I see some issues with the current approach: 1. If we add some new delegatee then it's impossible to distinguish for which delegatee given backend can be delegated. For example, we add new delegating backend with extended logging. How RocksDbBackend can signal that it can NOT work with it? 2. Every new backend must be marked explicitly ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java ########## @@ -204,6 +222,11 @@ public static StateBackend fromApplicationOrConfigOrDefault( // (1) the application defined state backend has precedence if (fromApplication != null) { + + checkArgument( + !(fromApplication instanceof DelegateStateBackend), + "DelegateStateBackend can not be delegated!"); Review comment: Is the intention here to prevent wrapping the backend twice? If so, I think such a check should be placed just before the actual wrapping. Another concern: can we get `DelegateStateBackend` here after deserialization? ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/delegate/DelegateStateBackend.java ########## @@ -0,0 +1,43 @@ +/* + * 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.apache.flink.runtime.state.delegate; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.state.StateBackend; + +/** + * An abstract base class for delegated state backend. + * + * <p>As its name, it should include a state backend to delegate, which can be one of the following: + * {@code MemoryStateBackend}, {code FsStateBackend} and {@code RocksDBStateBackend}. + */ +@Internal +public abstract class DelegateStateBackend implements StateBackend, java.io.Serializable { Review comment: Could you explain why do we need this class? ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/CheckpointStorageLoader.java ########## @@ -162,14 +163,19 @@ public static CheckpointStorage load( Preconditions.checkNotNull(classLoader, "classLoader"); Preconditions.checkNotNull(configuredStateBackend, "statebackend"); - if (configuredStateBackend instanceof CheckpointStorage) { + StateBackend rootStateBackend = + (configuredStateBackend instanceof DelegateStateBackend) + ? ((DelegateStateBackend) configuredStateBackend).getDelegatedStateBackend() + : configuredStateBackend; Review comment: This change will duplicated once we add some other delegating/proxying backend. Can we avoid it somehow? ########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/StateBackendLoader.java ########## @@ -242,6 +265,53 @@ public static StateBackend fromApplicationOrConfigOrDefault( return backend; } + public static StateBackend loadStateBackend( + @Nullable StateBackend fromApplication, + Configuration config, + ClassLoader classLoader, + @Nullable Logger logger) + throws IllegalConfigurationException, DynamicCodeLoadingException, IOException { + + final StateBackend backend = + fromApplicationOrConfigOrDefault(fromApplication, config, classLoader, logger); + + if (config.get(CheckpointingOptions.ENABLE_STATE_CHANGE_LOG)) { + + Preconditions.checkArgument( + backend instanceof DelegatedStateBackend, "backend is not delegable"); + + LOG.info( + "Delegate State Backend is used, and the root State Backend is {}", + backend.getClass().getSimpleName()); + + // ChangelogStateBackend resides in a separate module, load it using reflection + try { + Constructor<? extends DelegateStateBackend> constructor = + Class.forName(CHANGELOG_STATE_BACKEND, false, classLoader) + .asSubclass(DelegateStateBackend.class) + .getConstructor(DelegatedStateBackend.class); + + Class.forName(ROCKSDB_STATE_BACKEND_FACTORY, false, classLoader); Review comment: Could you add a comment why is it needed? ########## File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/changelog/state/ChangelogAggregatingState.java ########## @@ -0,0 +1,118 @@ +/* + * 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.apache.flink.changelog.state; Review comment: `package org.apache.flink.state.changelog;` ? ---------------------------------------------------------------- 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: us...@infra.apache.org
