[hotfix][config] Add env.isUnalignedCheckpointsEnabled() (java/scala/…

…python)

The next commits adds a force option, so this method is needed for
consistency.

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -771,3 +771,9 @@ def _generate_stream_graph(self, clear_transformations: bool = False, job_name:
  j_stream_graph.setJobName(job_name)
 
  return j_stream_graph
+
+ def is_unaligned_checkpoints_enabled(self):
+ """
+ Returns whether Unaligned Checkpoints are enabled.
+ """
+ return self._j_stream_execution_environment.isUnalignedCheckpointsEnabled()

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java

@@ -524,6 +524,14 @@ public boolean isForceCheckpointing() {
  return checkpointCfg.isForceCheckpointing();
  }
 
+ /**
+ * Returns whether Unaligned Checkpoints are enabled.
+ */
+ @PublicEvolving
+ public boolean isUnalignedCheckpointsEnabled() {
+ return checkpointCfg.isUnalignedCheckpointsEnabled();
+ }
+
  /**
  * Returns the checkpointing mode (exactly-once vs. at-least-once).
  *

flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala

@@ -881,6 +881,11 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
  def registerCachedFile(filePath: String, name: String, executable: Boolean): Unit = {
  javaEnv.registerCachedFile(filePath, name, executable)
  }
+
+ /**
+ * Returns whether Unaligned Checkpoints are enabled.
+ */
+ def isUnalignedCheckpointsEnabled: Boolean = javaEnv.isUnalignedCheckpointsEnabled
 }
 
 object StreamExecutionEnvironment {