**Apache Airflow version**: 2.0.1

**Environment**:

- **Cloud provider or hardware configuration**: AWS

**What happened**:
I have Airflow with Cloudwatch-based remote logging running. I also have `BashOperator` that does, for example, `rsync` with invalid parameters, for example `rsync -av test test`. The output of the `rsync` error is formatted and contains empty line. Once that empty line is logged to the Cloudwatch, i receive an error:
```
2021-04-06 19:29:22,318] /home/airflow/.local/lib/python3.6/site-packages/watchtower/__init__.py:154 WatchtowerWarning: Failed to deliver logs: Parameter validation failed:
Invalid length for parameter logEvents[5].message, value: 0, valid range: 1-inf
[2021-04-06 19:29:22,320] /home/airflow/.local/lib/python3.6/site-packages/watchtower/__init__.py:158 WatchtowerWarning: Failed to deliver logs: None
```

So basically empty lines can't be submitted to the Cloudwatch and as result the whole output of the process doesn't appear in logs.

**What you expected to happen**:

I expect to have an output of the bash command in logs. Empty lines can be skipped or replaced with something.

**How to reproduce it**:

For example: run `BashOperator` with `rsync` command that fails on Airflow with Cloudwatch-based remote logging. It could be any other command that produces empty line in the output.
airflow/providers/amazon/aws/log/cloudwatch_task_handler.py
return super()._render_filename(ti, try_number).replace(':', '_')
    def set_context(self, ti):
        super().set_context(ti)
        self.handler = watchtower.CloudWatchLogHandler(
            log_group=self.log_group,
            stream_name=self._render_filename(ti, ti.try_number),
            boto3_session=self.hook.get_session(self.region_name),
        )
    def close(self):
        """Close the handler responsible for the upload of the local log file to Cloudwatch."""
        
        
        
        
        if self.closed:
            return
        if self.handler is not None:
            self.handler.close()
        
        self.closed = True
    def _read(self, task_instance, try_number, metadata=None):