[FLINK-14068][core] Updates python sources

Author: XComp

flink-python/pyflink/fn_execution/embedded/java_utils.py

@@ -40,7 +40,6 @@
 
 # Java StateTtlConfig
 JStateTtlConfig = findClass('org.apache.flink.api.common.state.StateTtlConfig')
-JTime = findClass('java.time.Duration')
 JUpdateType = findClass('org.apache.flink.api.common.state.StateTtlConfig$UpdateType')
 JStateVisibility = findClass('org.apache.flink.api.common.state.StateTtlConfig$StateVisibility')
 
@@ -140,7 +139,7 @@ def to_java_typeinfo(type_info: TypeInformation):
 
 def to_java_state_ttl_config(ttl_config: StateTtlConfig):
     j_ttl_config_builder = JStateTtlConfig.newBuilder(
-        JTime.milliseconds(ttl_config.get_ttl().to_milliseconds()))
+        ttl_config.get_time_to_live())
 
     update_type = ttl_config.get_update_type()
     if update_type == StateTtlConfig.UpdateType.Disabled:

flink-python/pyflink/fn_execution/tests/test_flink_fn_execution_pb2.py

@@ -59,7 +59,7 @@ def test_state_ttl_config_proto(self):
             .build()
         state_ttl_config_proto = state_ttl_config._to_proto()
         state_ttl_config = StateTtlConfig._from_proto(state_ttl_config_proto)
-        self.assertEqual(state_ttl_config.get_ttl(), Time.milliseconds(1000))
+        self.assertEqual(state_ttl_config.get_time_to_live().toMillis(), 1000)
         self.assertEqual(
             state_ttl_config.get_update_type(), StateTtlConfig.UpdateType.OnCreateAndWrite)
         self.assertEqual(

flink-python/pyflink/table/table_config.py

@@ -191,8 +191,8 @@ def set_idle_state_retention_time(self,
                          0 (zero) to never clean-up the state.
         """
         j_time_class = get_gateway().jvm.java.time.Duration
-        j_min_time = j_time_class.milliseconds(long(round(min_time.total_seconds() * 1000)))
-        j_max_time = j_time_class.milliseconds(long(round(max_time.total_seconds() * 1000)))
+        j_min_time = j_time_class.ofMillis(long(round(min_time.total_seconds() * 1000)))
+        j_max_time = j_time_class.ofMillis(long(round(max_time.total_seconds() * 1000)))
         self._j_table_config.setIdleStateRetentionTime(j_min_time, j_max_time)
 
     def set_idle_state_retention(self, duration: datetime.timedelta):

flink-python/pyflink/util/java_utils.py

@@ -40,15 +40,14 @@ def to_jarray(j_type, arr):
 
 def to_j_flink_time(time_delta):
     gateway = get_gateway()
-    TimeUnit = gateway.jvm.java.util.concurrent.TimeUnit
-    Time = gateway.jvm.java.time.Duration
+    Duration = gateway.jvm.java.time.Duration
     if isinstance(time_delta, timedelta):
         total_microseconds = round(time_delta.total_seconds() * 1000 * 1000)
-        return Time.of(total_microseconds, TimeUnit.MICROSECONDS)
+        return Duration.ofMillis(total_microseconds)
     else:
         # time delta in milliseconds
         total_milliseconds = time_delta
-        return Time.milliseconds(total_milliseconds)
+        return Duration.ofMillis(total_milliseconds)
 
 
 def from_j_flink_time(j_flink_time):