### 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)

airflow/jobs/triggerer_job.py
"""
        Wrapper which runs an actual trigger (they are async generators)
        and pushes their events into our outbound event deque.
        """
        self.log.info("Trigger %s starting", self.triggers[trigger_id]['name'])
        try:
            async for event in trigger.run():
                self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]['name'], event)
                self.triggers[trigger_id]["events"] += 1
                self.events.append((trigger_id, event))
        finally:
            
            
            
            
            trigger.cleanup()
    
    def update_triggers(self, requested_trigger_ids: Set[int]):