KarmaGYZ commented on a change in pull request #16405: URL: https://github.com/apache/flink/pull/16405#discussion_r665000412
########## File path: flink-python/pyflink/datastream/slot_sharing_group.py ########## @@ -0,0 +1,282 @@ +################################################################################ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +################################################################################ + +__all__ = ['MemorySize', 'SlotSharingGroup'] + +from typing import Optional + +from pyflink.java_gateway import get_gateway + + +class MemorySize(object): + """ + MemorySize is a representation of a number of bytes, viewable in different units. + """ + + def __init__(self, j_memory_size=None, bytes_size: int = None): + self._j_memory_size = get_gateway().jvm \ + .org.apache.flink.configuration.MemorySize(bytes_size) \ + if j_memory_size is None else j_memory_size + + @staticmethod + def of_mebi_bytes(mebi_bytes: int) -> 'MemorySize': + return MemorySize( + get_gateway().jvm.org.apache.flink.configuration.MemorySize.ofMebiBytes(mebi_bytes)) + + def get_bytes(self) -> int: + """ + Gets the memory size in bytes. + + :return: The memory size in bytes. + """ + return self._j_memory_size.getBytes() + + def get_kibi_bytes(self) -> int: + """ + Gets the memory size in Kibibytes (= 1024 bytes). + + :return: The memory size in Kibibytes. + """ + return self._j_memory_size.getKibiBytes() + + def get_mebi_bytes(self) -> int: + """ + Gets the memory size in Mebibytes (= 1024 Kibibytes). + + :return: The memory size in Mebibytes. + """ + return self._j_memory_size.getMebiBytes() + + def get_gibi_bytes(self) -> int: + """ + Gets the memory size in Gibibytes (= 1024 Mebibytes). + + :return: The memory size in Gibibytes. + """ + return self._j_memory_size.getGibiBytes() + + def get_tebi_bytes(self) -> int: + """ + Gets the memory size in Tebibytes (= 1024 Gibibytes). + + :return: The memory size in Tebibytes. + """ + return self._j_memory_size.getTebiBytes() + + def get_java_memory_size(self): + """ + Get the Java MemorySize object. + + :return: The Java MemorySize object. + """ + return self._j_memory_size + + def __eq__(self, other): + return isinstance(other, self.__class__) and \ + self._j_memory_size == other._j_memory_size + + def __hash__(self): + return self._j_memory_size.hashCode() + + +class SlotSharingGroup(object): + """ + Describe the name and the the different resource components of a slot sharing group. + """ + + def __init__(self, j_slot_sharing_group): + self._j_slot_sharing_group = j_slot_sharing_group + + def get_name(self) -> str: + """ + Get the name of this SlotSharingGroup. + + :return: The name of the SlotSharingGroup. + """ + return self._j_slot_sharing_group.getName() + + def get_managed_memory(self) -> Optional[MemorySize]: + """ + Get the task managed memory for this SlotSharingGroup. + + :return: The task managed memory of the SlotSharingGroup. + """ + managed_memory = self._j_slot_sharing_group.getManagedMemory() + return MemorySize(managed_memory.get()) if managed_memory.isPresent() else None + + def get_task_heap_memory(self) -> Optional[MemorySize]: + """ + Get the task heap memory for this SlotSharingGroup. + + :return: The task heap memory of the SlotSharingGroup. + """ + task_heap_memory = self._j_slot_sharing_group.getTaskHeapMemory() + return MemorySize(task_heap_memory.get()) if task_heap_memory.isPresent() else None + + def get_task_off_heap_memory(self) -> Optional[MemorySize]: + """ + Get the task off-heap memory for this SlotSharingGroup. + + :return: The task off-heap memory of the SlotSharingGroup. + """ + task_off_heap_memory = self._j_slot_sharing_group.getTaskOffHeapMemory() + return MemorySize(task_off_heap_memory.get()) if task_off_heap_memory.isPresent() else None + + def get_cpu_cores(self) -> Optional[float]: + """ + G et the CPU cores for this SlotSharingGroup. Review comment: ```suggestion Get the CPU cores for this SlotSharingGroup. ``` -- 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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
