code: def test_file_sink(self):
      base_path = "/tmp/1.txt"
      encoder = Encoder.simple_string_encoder()
      file_sink_builder = FileSink.for_row_format(base_path, encoder)
      file_sink = (
          file_sink_builder.with_bucket_check_interval(1000)
          .with_bucket_assigner(
              BucketAssigner.base_path_bucket_assigner()
          )
          .with_rolling_policy(
              RollingPolicy.on_checkpoint_rolling_policy()
          )
          .with_output_file_config(
              OutputFileConfig.builder()
              .with_part_prefix("pre")
              .with_part_suffix("suf")
              .build()
          )
          .enable_compact(
              FileCompactStrategy.builder()
              .enable_compaction_on_checkpoint(3)
              .set_size_threshold(1024)
              .set_num_compact_threads(2)
              .build(),
              FileCompactor.concat_file_compactor(b"\n"),
          )
          .build()
      )
  
      buckets_builder_field = load_java_class(
          "org.apache.flink.connector.file.sink.FileSink"
      ).getDeclaredField("bucketsBuilder")
      buckets_builder_field.setAccessible(True)
      buckets_builder = buckets_builder_field.get(
          file_sink.get_java_function()
      )
  
      self.assertEqual(
          "DefaultRowFormatBuilder",
          buckets_builder.getClass().getSimpleName(),
      )
  
      row_format_builder_clz = load_java_class(
          "org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder"
      )
      encoder_field = row_format_builder_clz.getDeclaredField("encoder")
      encoder_field.setAccessible(True)
      self.assertEqual(
          "SimpleStringEncoder",
          encoder_field.get(buckets_builder).getClass().getSimpleName(),
      )
  
      interval_field = row_format_builder_clz.getDeclaredField(
          "bucketCheckInterval"
      )
      interval_field.setAccessible(True)
      self.assertEqual(1000, interval_field.get(buckets_builder))
  
      bucket_assigner_field = row_format_builder_clz.getDeclaredField(
          "bucketAssigner"
      )
      bucket_assigner_field.setAccessible(True)
      self.assertEqual(
          "BasePathBucketAssigner",
          bucket_assigner_field.get(buckets_builder)
          .getClass()
          .getSimpleName(),
      )
  
      rolling_policy_field = row_format_builder_clz.getDeclaredField(
          "rollingPolicy"
      )
      rolling_policy_field.setAccessible(True)
      self.assertEqual(
          "OnCheckpointRollingPolicy",
          rolling_policy_field.get(buckets_builder)
          .getClass()
          .getSimpleName(),
      )
  
      output_file_config_field = (
          row_format_builder_clz.getDeclaredField("outputFileConfig")
      )
      output_file_config_field.setAccessible(True)
      output_file_config = output_file_config_field.get(buckets_builder)
      self.assertEqual("pre", output_file_config.getPartPrefix())
      self.assertEqual("suf", output_file_config.getPartSuffix())
  
      compact_strategy_field = row_format_builder_clz.getDeclaredField(
          "compactStrategy"
      )
      compact_strategy_field.setAccessible(True)
      compact_strategy = compact_strategy_field.get(buckets_builder)
      self.assertEqual(
          3, compact_strategy.getNumCheckpointsBeforeCompaction()
      )
      self.assertEqual(1024, compact_strategy.getSizeThreshold())
      self.assertEqual(2, compact_strategy.getNumCompactThreads())
  
      file_compactor_field = row_format_builder_clz.getDeclaredField(
          "fileCompactor"
      )
      file_compactor_field.setAccessible(True)
      file_compactor = file_compactor_field.get(buckets_builder)
      self.assertEqual(
          "ConcatFileCompactor",
          file_compactor.getClass().getSimpleName(),
      )
      concat_file_compactor_clz = load_java_class(
          "org.apache.flink.connector.file.sink.compactor.ConcatFileCompactor"
      )
      file_delimiter_field = concat_file_compactor_clz.getDeclaredField(
          "fileDelimiter"
      )
      file_delimiter_field.setAccessible(True)
      file_delimiter = file_delimiter_field.get(file_compactor)
      self.assertEqual(b"\n", file_delimiter)
  
file_path: flink/flink-python/pyflink/datastream/connectors/tests/test_file_system.py
imports: from pyflink.common import Duration
  from pyflink.common.serialization import Encoder
  from pyflink.common.typeinfo import Types
  from pyflink.datastream.connectors.file_system import FileCompactStrategy, FileCompactor, StreamingFileSink, OutputFileConfig, FileSource, StreamFormat, FileEnumeratorProvider, FileSplitAssignerProvider, RollingPolicy, FileSink, BucketAssigner
  from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase
  from pyflink.util.java_utils import load_java_class
label_desc: Use image node_id 3 for calling the FileSystemTests obj's underlying member method code with example usage: obj.test_file_sink() without return types