dianfu commented on a change in pull request #12370:
URL: https://github.com/apache/flink/pull/12370#discussion_r433644752
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/CommonPythonBase.scala
##########
@@ -150,6 +152,64 @@ trait CommonPythonBase {
}
realEnv
}
+
+ private def isPythonWorkerUsingManagedMemory(config: Configuration): Boolean
= {
+ val clazz = loadClass("org.apache.flink.python.PythonOptions")
+ config.getBoolean(clazz.getField("USE_MANAGED_MEMORY").get(null)
+ .asInstanceOf[ConfigOption[java.lang.Boolean]])
+ }
+
+ private def getPythonWorkerMemory(config: Configuration): MemorySize = {
+ val clazz = loadClass("org.apache.flink.python.PythonOptions")
+ val pythonFrameworkMemorySize = MemorySize.parse(
+ config.getString(
+ clazz.getField("PYTHON_FRAMEWORK_MEMORY_SIZE").get(null)
+ .asInstanceOf[ConfigOption[String]]))
+ val pythonBufferMemorySize = MemorySize.parse(
+ config.getString(
+ clazz.getField("PYTHON_DATA_BUFFER_MEMORY_SIZE").get(null)
+ .asInstanceOf[ConfigOption[String]]))
+ pythonFrameworkMemorySize.add(pythonBufferMemorySize)
+ }
+
+ private def checkPythonWorkerMemory(
+ config: Configuration, env: StreamExecutionEnvironment = null): Unit = {
+ if (!isPythonWorkerUsingManagedMemory(config)) {
+ val taskOffHeapMemory =
config.get(TaskManagerOptions.TASK_OFF_HEAP_MEMORY)
+ val requiredPythonWorkerOffHeapMemory = getPythonWorkerMemory(config)
+ if (taskOffHeapMemory.compareTo(requiredPythonWorkerOffHeapMemory) < 0) {
+ throw new TableException(String.format("The configured Task Off-Heap
Memory %s is less " +
+ "than the least required Python worker Memory %s. The Task Off-Heap
Memory can be " +
+ "configured using the configuration key
'taskmanager.memory.task.off-heap.size'.",
+ taskOffHeapMemory, requiredPythonWorkerOffHeapMemory))
+ }
+ } else if (env != null && isRocksDbUsingManagedMemory(env)) {
+ throw new TableException("Currently it doesn't support to use Managed
Memory for both " +
+ "RocksDB state backend and Python worker at the same time. You can
either configure " +
+ "RocksDB state backend to use Task Off-Heap Memory via the
configuration key " +
+ "'state.backend.rocksdb.memory.managed' or configure Python worker to
use " +
+ "Task Off-Heap Memory via the configuration key " +
+ "'python.fn-execution.memory.managed'.")
+ }
+ }
+
+ private def isRocksDbUsingManagedMemory(env: StreamExecutionEnvironment):
Boolean = {
Review comment:
Do you mean the duplicate methods between the old planner and the blink
planner? If so, I'm afraid it's currently not possible as there is no way to
share code between them(this is also by design as we are not planning to
maintain two planners in the long term). Although it allows to share the code
technically by moving the shared code to flink-table-common, I think it's not a
good idea to do that for this case.
----------------------------------------------------------------
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
