[FLINK-24199][python] Expose StreamExecutionEnvironment#configure in Python API

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -22,7 +22,7 @@
 
 from py4j.java_gateway import JavaObject
 
-from pyflink.common import WatermarkStrategy
+from pyflink.common import Configuration, WatermarkStrategy
 from pyflink.common.execution_config import ExecutionConfig
 from pyflink.common.job_client import JobClient
 from pyflink.common.job_execution_result import JobExecutionResult
@@ -462,6 +462,25 @@ def get_stream_time_characteristic(self) -> 'TimeCharacteristic':
  j_characteristic = self._j_stream_execution_environment.getStreamTimeCharacteristic()
  return TimeCharacteristic._from_j_time_characteristic(j_characteristic)
 
+ def configure(self, configuration: Configuration):
+ """
+ Sets all relevant options contained in the :class:`~pyflink.common.Configuration`. such as
+ e.g. `pipeline.time-characteristic`. It will reconfigure
+ :class:`~pyflink.datastream.StreamExecutionEnvironment`,
+ :class:`~pyflink.common.ExecutionConfig` and :class:`~pyflink.datastream.CheckpointConfig`.
+
+ It will change the value of a setting only if a corresponding option was set in the
+ `configuration`. If a key is not present, the current value of a field will remain
+ untouched.
+
+ :param configuration: a configuration to read the values from.
+
+ .. versionadded:: 1.15.0
+ """
+ self._j_stream_execution_environment.configure(configuration._j_configuration,
+ get_gateway().jvm.Thread.currentThread()
+ .getContextClassLoader())
+
  def add_python_file(self, file_path: str):
  """
  Adds a python dependency which could be python files, python packages or

flink-python/pyflink/datastream/tests/test_stream_execution_environment.py

@@ -26,7 +26,7 @@
 import unittest
 import uuid
 
-from pyflink.common import ExecutionConfig, RestartStrategies
+from pyflink.common import Configuration, ExecutionConfig, RestartStrategies
 from pyflink.common.serialization import JsonRowDeserializationSchema
 from pyflink.common.typeinfo import Types
 from pyflink.datastream import (StreamExecutionEnvironment, CheckpointConfig,
@@ -195,6 +195,21 @@ def test_get_set_stream_time_characteristic(self):
 
  self.assertEqual(time_characteristic, TimeCharacteristic.ProcessingTime)
 
+ def test_configure(self):
+ configuration = Configuration()
+ configuration.set_string('pipeline.operator-chaining', 'false')
+ configuration.set_string('pipeline.time-characteristic', 'IngestionTime')
+ configuration.set_string('execution.buffer-timeout', '1 min')
+ configuration.set_string('execution.checkpointing.timeout', '12000')
+ configuration.set_string('state.backend', 'jobmanager')
+ self.env.configure(configuration)
+ self.assertEqual(self.env.is_chaining_enabled(), False)
+ self.assertEqual(self.env.get_stream_time_characteristic(),
+ TimeCharacteristic.IngestionTime)
+ self.assertEqual(self.env.get_buffer_timeout(), 60000)
+ self.assertEqual(self.env.get_checkpoint_config().get_checkpoint_timeout(), 12000)
+ self.assertTrue(isinstance(self.env.get_state_backend(), MemoryStateBackend))
+
  @unittest.skip("Python API does not support DataStream now. refactor this test later")
  def test_get_execution_plan(self):
  tmp_dir = tempfile.gettempdir()

flink-python/pyflink/datastream/tests/test_stream_execution_environment_completeness.py

@@ -47,7 +47,7 @@ def excluded_methods(cls):
  'readFileStream', 'isForceCheckpointing', 'readFile', 'clean',
  'createInput', 'createLocalEnvironmentWithWebUI', 'fromCollection',
  'socketTextStream', 'initializeContextEnvironment', 'readTextFile',
- 'setNumberOfExecutionRetries', 'configure', 'executeAsync', 'registerJobListener',
+ 'setNumberOfExecutionRetries', 'executeAsync', 'registerJobListener',
  'clearJobListeners', 'getJobListeners', 'fromSequence', 'getConfiguration',
  'generateStreamGraph'}