Issue 1:
Description

we need a feature where we can set the state of the task instance to the specified state.

Use case / motivation

In my application, some task instances have been scheduled for a long time, we want to put it in the failed state. Currently, we are doing this by using airflow Web UI but we want to use API instead.

Are you willing to submit a PR?

Yes, I'm willing to.

Related Issues

#16412

Issue 2:
Apache Airflow version: 2.0.2

Kubernetes version (if you are using kubernetes) (use kubectl version):

Environment:

Cloud provider or hardware configuration:

OS (e.g. from /etc/os-release):

Kernel (e.g. uname -a):

Install tools:

Others:

What happened:

ssh-keygen -t rsa -P "" -f test_rsa
cat test_rsa | python -c 'import sys;import json; print(json.dumps(sys.stdin.read()))' # gives the private key encoded as JSON string to be pasted in the connection extra private_key

I created an Airflow Connection

type: ssh

extra:

{
"look_for_keys": "false",
"no_host_key_check": "true",
"private_key": "-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn\nNhAA........W4tTGFndW5hLU1hY0Jvb2stUHJvAQI=\n-----END OPENSSH PRIVATE KEY-----\n",
"private_key_passphrase": ""
}

When this SSH connection is used in SFTPToS3Operator for example it will incorrectly parse that private_key as a paramiko.dsskey.DSSKey instead of the correct paramiko.rsakey.RSAKey.

The code responsible for the processing of private_key is not not deterministic (I don't think .values() returns items in any particular order) , but in my case it will always try paramiko.dsskey.DSSKey before it tries paramiko.rsakey.RSAKey:

airflow/airflow/providers/ssh/hooks/ssh.py

Lines 363 to 369
in
8e2a0bc

allowed_pkey_types = self._default_pkey_mappings.values()

for pkey_type in allowed_pkey_types:

try:

key = pkey_type.from_private_key(StringIO(private_key), password=passphrase)

return key

except paramiko.ssh_exception.SSHException:

continue

This incorrectly parsed private key will cause a very confusing error later when it's actually used

[2021-06-30 23:33:14,604] {transport.py:1819} INFO - Connected (version 2.0, client AWS_SFTP_1.0)
[2021-06-30 23:33:14,732] {transport.py:1819} ERROR - Unknown exception: q must be exactly 160, 224, or 256 bits long
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR - Traceback (most recent call last):
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/transport.py", line 2109, in run
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR - handler(self.auth_handler, m)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/auth_handler.py", line 298, in _parse_service_accept
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - sig = self.private_key.sign_ssh_data(blob)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/dsskey.py", line 108, in sign_ssh_data
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - key = dsa.DSAPrivateNumbers(
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py", line 244, in private_key
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - return backend.load_dsa_private_numbers(self)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/backends/openssl/backend.py", line 826, in load_dsa_private_numbers
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR - dsa._check_dsa_private_numbers(numbers)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py", line 282, in _check_dsa_private_numbers
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - _check_dsa_parameters(parameters)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py", line 274, in _check_dsa_parameters
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - raise ValueError("q must be exactly 160, 224, or 256 bits long")
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - ValueError: q must be exactly 160, 224, or 256 bits long
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -

What you expected to happen:

I expected to parse the private_key as a RSAKey.

I did my own test and paramiko.dsskey.DSSKey.from_private_key(StringIO(private_key), password=passphrase) will happily parse (incorrectly) a RSA key. The current code assumes that it will raise an exception but it won't.

How to reproduce it:

Anything else we need to know:

For me it happens always, I don't think the order of .values() is deterministic, but in my laptop it will always try DSSKey before RSAKey.