### Description

Add feature to EmrStepSensor to bring back the spark task url & logs after task execution

### Use case/motivation

After starting an EMR step task using EmrAddStepsOperator we generally have an EmrStepSensor to track the status of the step. The job ID is available for the sensor and is being poked at regular interval. 

```
[2022-04-26, 22:07:43 UTC] {base_aws.py:100} INFO - Retrieving region_name from Connection.extra_config['region_name']
[2022-04-26, 22:07:44 UTC] {emr.py:316} INFO - Poking step s-123ABC123ABC on cluster j-123ABC123ABC
[2022-04-26, 22:07:44 UTC] {emr.py:74} INFO - Job flow currently PENDING
[2022-04-26, 22:08:44 UTC] {emr.py:316} INFO - Poking step s-123ABC123ABC on cluster j-123ABC123ABC
[2022-04-26, 22:08:44 UTC] {emr.py:74} INFO - Job flow currently PENDING
[2022-04-26, 22:09:44 UTC] {emr.py:316} INFO - Poking step s-123ABC123ABC on cluster j-123ABC123ABC
[2022-04-26, 22:09:44 UTC] {emr.py:74} INFO - Job flow currently COMPLETED
[2022-04-26, 22:09:44 UTC] {base.py:251} INFO - Success criteria met. Exiting.
[2022-04-26, 22:09:44 UTC] {taskinstance.py:1288} INFO - Marking task as SUCCESS. dag_id=datapipeline_sample, task_id=calculate_pi_watch_step, execution_date=20220426T220739, start_date=20220426T220743, end_date=20220426T220944
```

After the task is completed the status is displayed. If the user wants to review the logs of the task, it is a multistep process to get hold of the job logs from EMR cluster. 

It will be a great addition to add the log url and possibly relay the logs to Airflow EmrStepSensor post completion of the task. This will be very handy when there are failures of many tasks and will make it a great user experience. 


### Related issues

_No response_

### Are you willing to submit a 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)

# asse
        mock_hook = S3Hook()
        mock_hook.create_bucket(bucket_name=s3_bucket)
        assert mock_hook.check_for_bucket(bucket_name=s3_bucket)
        mock_hook.delete_bucket(bucket_name=s3_bucket, force_delete=True)
        assert not mock_hook.check_for_bucket(s3_bucket)
    @mock_s3
    def test_delete_bucket_if_not_bucket_exist(self, s3_bucket):
        # asse
        mock_hook = S3Hook()
        with pytest.raises(ClientError) as ctx:
            assert mock_hook.delete_bucket(bucket_name=s3_bucket, force_delete=True)
        assert ctx.value.response["Error"]["Code"] == "NoSuchBucket"
    @mock.patch.object(S3Hook, "get_connection", return_value=Connection(schema="test_bucket"))
    def test_provide_bucket_name(self, mock_get_connection):
        class FakeS3Hook(S3Hook):
            @provide_bucket_name
            def test_function(self, bucket_name=None):
                return bucket_name
        fake_s3_hook = FakeS3Hook()
        test_bucket_name = fake_s3_hook.test_function()
        assert test_bucket_name == mock_get_connection.return_value.schema
        test_bucket_name = fake_s3_hook.test_function(bucket_name="bucket")
        assert test_bucket_name == "bucket"
    def test_delete_objects_key_does_not_exist(self, s3_bucket):
        # The 
        # It w
        hook = S3Hook()
        hook.delete_objects(bucket=s3_bucket, keys=["key-1"])
    def test_delete_objects_one_key(self, mocked_s3_res, s3_bucket):