class_docstr: **IMPORTANT** `FsStateBackend is deprecated in favor of `HashMapStateBackend` and
  `FileSystemCheckpointStorage`. This change does not affect the runtime characteristics
  of your Jobs and is simply an API change to help better communicate the ways Flink separates
  local state storage from fault tolerance. Jobs can be upgraded without loss of state. If
  configuring your state backend via the `StreamExecutionEnvironment` please make the following
  changes.
  
  ::
  
      >> env.set_state_backend(HashMapStateBackend())
      >> env.get_checkpoint_config().set_checkpoint_storage("hdfs://checkpoints")
  
  If you are configuring your state backend via the `flink-conf.yaml` please set your state
  backend type to `hashmap`.
  
  This state backend holds the working state in the memory (JVM heap) of the TaskManagers.
  The state backend checkpoints state as files to a file system (hence the backend's name).
  
  Each checkpoint individually will store all its files in a subdirectory that includes the
  checkpoint number, such as ``hdfs://namenode:port/flink-checkpoints/chk-17/``.
  
  **State Size Considerations**
  
  Working state is kept on the TaskManager heap. If a TaskManager executes multiple
  tasks concurrently (if the TaskManager has multiple slots, or if slot-sharing is used)
  then the aggregate state of all tasks needs to fit into that TaskManager's memory.
  
  This state backend stores small state chunks directly with the metadata, to avoid creating
  many small files. The threshold for that is configurable. When increasing this threshold, the
  size of the checkpoint metadata increases. The checkpoint metadata of all retained completed
  checkpoints needs to fit into the JobManager's heap memory. This is typically not a problem,
  unless the threshold :func:`get_min_file_size_threshold` is increased significantly.
  
  **Persistence Guarantees**
  
  Checkpoints from this state backend are as persistent and available as filesystem that is
  written to. If the file system is a persistent distributed file system, this state backend
  supports highly available setups. The backend additionally supports savepoints and externalized
  checkpoints.
  
  **Configuration**
  
  As for all state backends, this backend can either be configured within the application (by
  creating the backend with the respective constructor parameters and setting it on the execution
  environment) or by specifying it in the Flink configuration.
  
  If the state backend was specified in the application, it may pick up additional configuration
  parameters from the Flink configuration. For example, if the backend if configured in the
  application without a default savepoint directory, it will pick up a default savepoint
  directory specified in the Flink configuration of the running job/cluster. That behavior is
  implemented via the :func:`configure` method.
class_docstr_tok: ["*","*","IMPORTANT","*","*","`","FsStateBackend","is","deprecated","in","favor","of","`","HashMapStateBackend","`","and","`","FileSystemCheckpointStorage","`",".","This","change","does","not","affect","the","runtime","characteristics","of","your","Jobs","and","is","simply","an","API","change","to","help","better","communicate","the","ways","Flink","separates","local","state","storage","from","fault","tolerance",".","Jobs","can","be","upgraded","without","loss","of","state",".","If","configuring","your","state","backend","via","the","`","StreamExecutionEnvironment","`","please","make","the","following","changes",".",":",":",">",">","env.set_state_backend","(","HashMapStateBackend","(",")",")",">",">","env.get_checkpoint_config","(",")",".set_checkpoint_storage","(","``","hdfs",":","\/\/checkpoints","''",")","If","you","are","configuring","your","state","backend","via","the","`","flink-conf.yaml","`","please","set","your","state","backend","type","to","`","hashmap","`",".","This","state","backend","holds","the","working","state","in","the","memory","(","JVM","heap",")","of","the","TaskManagers",".","The","state","backend","checkpoints","state","as","files","to","a","file","system","(","hence","the","backend","'s","name",")",".","Each","checkpoint","individually","will","store","all","its","files","in","a","subdirectory","that","includes","the","checkpoint","number",",","such","as","``","hdfs",":","\/\/namenode",":","port\/flink-checkpoints\/chk-17\/","``",".","*","*","State","Size","Considerations","*","*","Working","state","is","kept","on","the","TaskManager","heap",".","If","a","TaskManager","executes","multiple","tasks","concurrently","(","if","the","TaskManager","has","multiple","slots",",","or","if","slot-sharing","is","used",")","then","the","aggregate","state","of","all","tasks","needs","to","fit","into","that","TaskManager","'s","memory",".","This","state","backend","stores","small","state","chunks","directly","with","the","metadata",",","to","avoid","creating","many","small","files",".","The","threshold","for","that","is","configurable",".","When","increasing","this","threshold",",","the","size","of","the","checkpoint","metadata","increases",".","The","checkpoint","metadata","of","all","retained","completed","checkpoints","needs","to","fit","into","the","JobManager","'s","heap","memory",".","This","is","typically","not","a","problem",",","unless","the","threshold",":","func",":","`","get_min_file_size_threshold","`","is","increased","significantly",".","*","*","Persistence","Guarantees","*","*","Checkpoints","from","this","state","backend","are","as","persistent","and","available","as","filesystem","that","is","written","to",".","If","the","file","system","is","a","persistent","distributed","file","system",",","this","state","backend","supports","highly","available","setups",".","The","backend","additionally","supports","savepoints","and","externalized","checkpoints",".","*","*","Configuration","*","*","As","for","all","state","backends",",","this","backend","can","either","be","configured","within","the","application","(","by","creating","the","backend","with","the","respective","constructor","parameters","and","setting","it","on","the","execution","environment",")","or","by","specifying","it","in","the","Flink","configuration",".","If","the","state","backend","was","specified","in","the","application",",","it","may","pick","up","additional","configuration","parameters","from","the","Flink","configuration",".","For","example",",","if","the","backend","if","configured","in","the","application","without","a","default","savepoint","directory",",","it","will","pick","up","a","default","savepoint","directory","specified","in","the","Flink","configuration","of","the","running","job\/cluster",".","That","behavior","is","implemented","via","the",":","func",":","`","configure","`","method","."]
docstr: Creates a new state backend that stores its checkpoint data in the file system and location
  defined by the given URI.
  
  A file system for the file system scheme in the URI (e.g., 'file://', 'hdfs://', or
  'S3://') must be accessible via ``org.apache.flink.core.fs.FileSystem.get(URI)``.
  
  For a state backend targeting HDFS, this means that the URI must either specify the
  authority (host and port), or that the Hadoop configuration that describes that information
  must be in the classpath.
  
  Example:
  ::
  
      >>> state_backend = FsStateBackend("file://var/checkpoints/")
  
  
  :param checkpoint_directory_uri: The path to write checkpoint metadata to, required.
  :param default_savepoint_directory_uri: The path to write savepoints to. If none, the value
                                          from the runtime configuration will be used, or
                                          savepoint target locations need to be passed when
                                          triggering a savepoint.
  :param file_state_size_threshold: State below this size will be stored as part of the
                                    metadata, rather than in files. If none, the value
                                    configured in the runtime configuration will be used, or
                                    the default value (1KB) if nothing is configured.
  :param write_buffer_size: Write buffer size used to serialize state. If -1, the value
                            configured in the runtime configuration will be used, or the
                            default value (4KB) if nothing is configured.
  :param using_asynchronous_snapshots: Snapshots are now always asynchronous. This flag
                                       has no effect anymore in this version.
  :param j_fs_state_backend: For internal use, please keep none.
code: def __init__(
      self,
      checkpoint_directory_uri=None,
      default_savepoint_directory_uri=None,
      file_state_size_threshold=None,
      write_buffer_size=None,
      using_asynchronous_snapshots=None,
      j_fs_state_backend=None,
  ):
      """
      Creates a new state backend that stores its checkpoint data in the file system and location
      defined by the given URI.
  
      A file system for the file system scheme in the URI (e.g., 'file://', 'hdfs://', or
      'S3://') must be accessible via ``org.apache.flink.core.fs.FileSystem.get(URI)``.
  
      For a state backend targeting HDFS, this means that the URI must either specify the
      authority (host and port), or that the Hadoop configuration that describes that information
      must be in the classpath.
  
      Example:
      ::
  
          >>> state_backend = FsStateBackend("file://var/checkpoints/")
  
  
      :param checkpoint_directory_uri: The path to write checkpoint metadata to, required.
      :param default_savepoint_directory_uri: The path to write savepoints to. If none, the value
                                              from the runtime configuration will be used, or
                                              savepoint target locations need to be passed when
                                              triggering a savepoint.
      :param file_state_size_threshold: State below this size will be stored as part of the
                                        metadata, rather than in files. If none, the value
                                        configured in the runtime configuration will be used, or
                                        the default value (1KB) if nothing is configured.
      :param write_buffer_size: Write buffer size used to serialize state. If -1, the value
                                configured in the runtime configuration will be used, or the
                                default value (4KB) if nothing is configured.
      :param using_asynchronous_snapshots: Snapshots are now always asynchronous. This flag
                                           has no effect anymore in this version.
      :param j_fs_state_backend: For internal use, please keep none.
      """
      if j_fs_state_backend is None:
          gateway = get_gateway()
          JTernaryBoolean = (
              gateway.jvm.org.apache.flink.util.TernaryBoolean
          )
          JFsStateBackend = (
              gateway.jvm.org.apache.flink.runtime.state.filesystem.FsStateBackend
          )
          JPath = gateway.jvm.org.apache.flink.core.fs.Path
          if checkpoint_directory_uri is None:
              raise ValueError(
                  "The parameter 'checkpoint_directory_uri' is required!"
              )
          j_checkpoint_directory_uri = JPath(
              checkpoint_directory_uri
          ).toUri()
  
          if default_savepoint_directory_uri is None:
              j_default_savepoint_directory_uri = None
          else:
              j_default_savepoint_directory_uri = JPath(
                  default_savepoint_directory_uri
              ).toUri()
  
          if file_state_size_threshold is None:
              file_state_size_threshold = -1
  
          if write_buffer_size is None:
              write_buffer_size = -1
  
          if using_asynchronous_snapshots is None:
              j_asynchronous_snapshots = JTernaryBoolean.UNDEFINED
          elif using_asynchronous_snapshots is True:
              j_asynchronous_snapshots = JTernaryBoolean.TRUE
          elif using_asynchronous_snapshots is False:
              j_asynchronous_snapshots = JTernaryBoolean.FALSE
          else:
              raise TypeError(
                  "Unsupported input for 'using_asynchronous_snapshots': %s, "
                  "the value of the parameter should be None or"
                  "True or False."
              )
  
          j_fs_state_backend = JFsStateBackend(
              j_checkpoint_directory_uri,
              j_default_savepoint_directory_uri,
              file_state_size_threshold,
              write_buffer_size,
              j_asynchronous_snapshots,
          )
  
      self._j_fs_state_backend = j_fs_state_backend
      super(FsStateBackend, self).__init__(j_fs_state_backend)
  
imports: from abc import ABCMeta
  from enum import Enum
  from py4j.java_gateway import get_java_class
  from typing import List, Optional
  from pyflink.java_gateway import get_gateway
  from pyflink.util.java_utils import load_java_class
label_desc: Use image node_id 1 to create a new FsStateBackend object from inherited base classes: StateBackend with example: obj = FsStateBackend(checkpoint_directory_uri, default_savepoint_directory_uri, file_state_size_threshold, write_buffer_size, using_asynchronous_snapshots, j_fs_state_backend)