[FLINK-37398][state] Add proper message when state processor API is used with changelog

Author: gaborgsomogyi

flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointTaskStateManager.java

@@ -124,7 +124,7 @@ public StateChangelogStorage<?> getStateChangelogStorage() {
     @Override
     public StateChangelogStorageView<?> getStateChangelogStorageView(
             Configuration configuration, ChangelogStateHandle changelogStateHandle) {
-        return null;
+        throw new UnsupportedOperationException("State processor api does not support changelog.");
     }
 
     @Nullable

flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/restore/ChangelogBackendRestoreOperation.java

@@ -27,6 +27,7 @@
 import org.apache.flink.runtime.state.changelog.ChangelogStateHandle;
 import org.apache.flink.runtime.state.changelog.StateChange;
 import org.apache.flink.runtime.state.changelog.StateChangelogHandleReader;
+import org.apache.flink.runtime.state.changelog.StateChangelogStorageView;
 import org.apache.flink.state.changelog.ChangelogKeyedStateBackend;
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.Preconditions;
@@ -39,6 +40,8 @@
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * Restores {@link ChangelogKeyedStateBackend} from the provided {@link ChangelogStateBackendHandle
  * handles}.
@@ -97,11 +100,11 @@ private static <T extends ChangelogStateHandle> void readBackendHandle(
         Map<Short, StateID> stateIds = new HashMap<>();
         for (ChangelogStateHandle changelogHandle :
                 backendHandle.getNonMaterializedStateHandles()) {
+            StateChangelogStorageView<?> stateChangelogStorageView =
+                    taskStateManager.getStateChangelogStorageView(configuration, changelogHandle);
+            requireNonNull(stateChangelogStorageView, "Changelog storage view must not be null.");
             StateChangelogHandleReader<T> changelogHandleReader =
-                    (StateChangelogHandleReader<T>)
-                            taskStateManager
-                                    .getStateChangelogStorageView(configuration, changelogHandle)
-                                    .createReader();
+                    (StateChangelogHandleReader<T>) stateChangelogStorageView.createReader();
             try (CloseableIterator<StateChange> changes =
                     changelogHandleReader.getChanges((T) changelogHandle)) {
                 while (changes.hasNext()) {