### Apache Airflow version

2.2.0b1 (beta snapshot)

### Operating System

debian buster

### Versions of Apache Airflow Providers

n/a

### Deployment

Astronomer

### Deployment details

Helm info:
```
helm repo add astronomer https://helm.astronomer.io
cat <<- 'EOF'  | helm install airflow astronomer/airflow --namespace airflow -f -
airflowVersion: 2.2.0
defaultAirflowRepository: myrepo
defaultAirflowTag: myimagetag
executor: KubernetesExecutor
images:
  airflow:
    repository: myrepo
    pullPolicy: Always
  pod_template:
    repository: myrepo
    pullPolicy: Always
triggerer:
  serviceAccount:
    create: True
EOF
```

Dockerfile:
```
FROM quay.io/astronomer/ap-airflow-dev:2.2.0-buster-43897
COPY ./dags/ ./dags/
```

[the dag](https://gist.github.com/MatrixManAtYrService/8d1d0c978465aa249e8bd0498cc08031#file-many_triggers-py)

### What happened

Six tasks deferred for a random (but deterministic) amount of time, and the triggerer fired six events.  Two of those events then deferred a second time, which wasn't necessary because they had already fired.  Looks like a race condition.

Here are the triggerer logs: https://gist.github.com/MatrixManAtYrService/8d1d0c978465aa249e8bd0498cc08031#file-triggerer-log-L29

Note that most deferrals only appear once, but the ones for "Aligator" and  "Bear" appear twice.

### What you expected to happen

Six tasks deferred, six tasks fires, no extra deferrals.

### How to reproduce

Run the dag in [this gist](https://gist.github.com/MatrixManAtYrService/8d1d0c978465aa249e8bd0498cc08031) with the kubernetes executor (be sure there's a triggerer running).  Notice that some of the triggerer log messages appear nore than once.  Those represent superfluous computation.

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

tests/jobs/test_triggerer_job.py
"""Checks the heartbeat logic"""
    
    triggerer_job = TriggererJob(None, heartrate=10, state=State.RUNNING)
    assert triggerer_job.is_alive()
    
    triggerer_job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=20)
    assert triggerer_job.is_alive()
    
    triggerer_job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=31)
    assert not triggerer_job.is_alive()
    
    triggerer_job.state = State.SUCCESS
    triggerer_job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
    assert not triggerer_job.is_alive(), "Completed jobs even with recent heartbeat should not be alive"
@pytest.mark.skipif(sys.version_info.minor <= 6 and sys.version_info.major <= 3, reason="No triggerer on 3.6")
def test_is_needed(session):
    """Checks the triggerer-is-needed logic"""
    
    triggerer_job = TriggererJob(None, heartrate=10, state=State.RUNNING)
    assert triggerer_job.is_needed() is False
    
    trigger = TimeDeltaTrigger(datetime.timedelta(days=7))
    trigger_orm = Trigger.from_object(trigger)
    trigger_orm.id = 1
    session.add(trigger_orm)
    session.commit()
    assert triggerer_job.is_needed() is True
@pytest.mark.skipif(sys.version_info.minor <= 6 and sys.version_info.major <= 3, reason="No triggerer on 3.6")
def test_capacity_decode():