[Bug] [FTP] The connections of SFTP didn't release when task failed.

[FlechazoW]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

When the task fails due to issues like 'no such file' or 'file is empty', the sftp connection didn't release.

Task mode is yarn-session.

What you expected to happen

Connection is released.

How to reproduce

1. Create a task which source is 'sftp'.
2. The file that need to be synced is empty or not exist.

{
  "job": {
    "content": [
      {
        "reader": {
          "parameter": {
            "path": "/root/tiezhu/test_two.csv",
            "protocol": "SFTP",
            "password": "Abc!@#135",
            "port": 22,
            "isFirstLineHeader": true,
            "host": "172.16.85.171",
            "column": [
              {
                "name": "id",
                "type": "int"
              },
              {
                "name": "name",
                "type": "string"
              }
            ],
            "fieldDelimiter": ",",
            "encoding": "utf-8",
            "fileType": "CSV",
            "username": "root"
          },
          "name": "ftpreader"
        },
        "writer": {
          "parameter": {
            "print": true
          },
          "name": "streamwriter"
        }
      }
    ]
  }
}

Anything else

NONE

Version

master

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[FlechazoW]

same bug with local test mode.

{
  "job": {
    "content": [
      {
        "reader": {
          "parameter": {
            "password": "DT@Stack#123",
            "column": [
              {
                "name": "id",
                "type": "INT"
              },
              {
                "name": "result_val",
                "type": "VARCHAR"
              }
            ],
            "connection": [
              {
                "schema": "flinkx_test",
                "jdbcUrl": [
                  "jdbc:mysql://172.16.100.186:3306/flinkx_test?useSSL=false"
                ],
                "table": [
                  "ftp_source"
                ]
              }
            ],
            "splitPk": "id",
            "username": "drpeco"
          },
          "name": "mysqlreader"
        },
        "writer": {
          "parameter": {
            "path": "/data/sftp/buffertest",
            "password": "dt@sz.com",
            "protocol": "SFTP",
            "port": "22",
            "host": "172.16.100.116",
            "column": [
              {
                "name": "id",
                "type": "INT"
              },
              {
                "name": "result_val",
                "type": "VARCHAR"
              }
            ],
            "writeMode": "append",
            "fieldDelimiter": ",",
            "ftpFileName": "xiaohe.txt",
            "username": "root"
          },
          "name": "ftpwriter",
          "table": {
            "tableName": "sink"
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "bytes": 0,
        "channel": 2
      }
    }
  }
}