env {
  execution.parallelism = 2
  job.mode = "BATCH"

}
source{
   SftpFile {
    path = "/home/dataos/test"
    host = "xxxx"
    port = 22
    user = xxx
    password = "****"
    file_format_type = "text"
    result_table_name = SftpFile_source_4
    schema = {
     fields {
      name = string
      age = int
      }
    }
    delimiter = ","
   }
  } 
  
 transform{
    Sql{
     source_table_name = "SftpFile_source_4"
    result_table_name = "Sql-transform-4"
    query = "select name,age, DES_ENCRYPT('1111111111',name)  des_name from SftpFile_source_4"
    }
  }
  
  sink {
    SftpFile{
      file_format_type="text",
      path = "/home/test1"
      host = "xxx"
      port = 22
      user = xxx
      password = "****"
      file_format_type = "text"
      field_delimiter=",",
      row_delimiter= "/t",
      sink_columns = ["name","age","des_name"]
      source_table_name="Sql-transform-4" 
    }
  }

1.Terminal shows that the SFTP connection is not released.
2.Debugging reveals that the file stream is not closed when the SFTPInputStream.close method is triggered.