Repository: PrefectHQ/prefect

Issue:
AzureResult read function not working
## Description
<!-- A clear description of the bug -->
Using `AzureResult` to read a file, an exception is raised:

> TypeError: a bytes-like object is required, not 'StorageStreamDownloader'

Looking at the `read` function in `src/prefect/engine/results/azure_result.py` I see the following:

```python
content_string = client.download_blob()

try:
    new.value = new.serializer.deserialize(content_string)
```

`client.download_blob()` returns a `StorageStreamDownloader`, while `new.serailizer.deserialize(content_string)` expects a byte array.

## Expected Behavior
I would expect that the file would be read without raising an exception.

## Reproduction
```python
from prefect.engine.results import AzureResult

con_string = ""
azure_result = AzureResult(container="data", connection_string=con_string)

w = azure_result.write({"Test": "123"})
r = azure_result.read(w.location)
```


## Environment
```json
{
  "config_overrides": {
    "context": {
      "secrets": false
    }
  },
  "env_vars": [],
  "system_information": {
    "platform": "Windows-10-10.0.18362-SP0",
    "prefect_backend": "server",
    "prefect_version": "0.13.19",
    "python_version": "3.9.0"
  }
}
```

Discussion:

Patch:
diff --git a/changes/issue3846.yaml b/changes/issue3846.yaml
new file mode 100644
index 0000000000..6462b652f4
--- /dev/null
+++ b/changes/issue3846.yaml
@@ -0,0 +1,2 @@
+fix:
+  - "Fix Azure result byte decoding of blob data - [#3846](https://github.com/PrefectHQ/prefect/issues/3846)"
diff --git a/changes/pr3842.yaml b/changes/pr3842.yaml
deleted file mode 100644
index 53065c4dc5..0000000000
--- a/changes/pr3842.yaml
+++ /dev/null
@@ -1,2 +0,0 @@
-enhancement:
-  - "Remove task run level heartbeats for performance - [#3842](https://github.com/PrefectHQ/prefect/pull/3842)"
diff --git a/src/prefect/engine/cloud/task_runner.py b/src/prefect/engine/cloud/task_runner.py
index 40c4f7edd3..858b73d0f5 100644
--- a/src/prefect/engine/cloud/task_runner.py
+++ b/src/prefect/engine/cloud/task_runner.py
@@ -20,6 +20,7 @@ from prefect.engine.state import (
 from prefect.engine.task_runner import TaskRunner, TaskRunnerInitializeResult
 from prefect.utilities.exceptions import VersionLockError
 from prefect.utilities.executors import tail_recursive
+from prefect.utilities.graphql import with_args
 
 
 class CloudTaskRunner(TaskRunner):
@@ -53,6 +54,31 @@ class CloudTaskRunner(TaskRunner):
             task=task, state_handlers=state_handlers, flow_result=flow_result
         )
 
+    def _heartbeat(self) -> bool:
+        try:
+            task_run_id = self.task_run_id  # type: str
+            self.heartbeat_cmd = ["prefect", "heartbeat", "task-run", "-i", task_run_id]
+            self.client.update_task_run_heartbeat(task_run_id)
+
+            # use empty string for testing purposes
+            flow_run_id = prefect.context.get("flow_run_id", "")  # type: str
+            query = {
+                "query": {
+                    with_args("flow_run_by_pk", {"id": flow_run_id}): {
+                        "flow": {"settings": True},
+                    }
+                }
+            }
+            flow_run = self.client.graphql(query).data.flow_run_by_pk
+            if not flow_run.flow.settings.get("heartbeat_enabled", True):
+                return False
+            return True
+        except Exception:
+            self.logger.exception(
+                "Heartbeat failed for Task '{}'".format(self.task.name)
+            )
+            return False
+
     def call_runner_target_handlers(self, old_state: State, new_state: State) -> State:
         """
         A special state handler that the TaskRunner uses to call its task's state handlers.
diff --git a/src/prefect/engine/results/azure_result.py b/src/prefect/engine/results/azure_result.py
index 876866ead0..1e8d3c8707 100644
--- a/src/prefect/engine/results/azure_result.py
+++ b/src/prefect/engine/results/azure_result.py
@@ -126,10 +126,10 @@ class AzureResult(Result):
             client = self.service.get_blob_client(
                 container=self.container, blob=location
             )
-            content_string = client.download_blob()
+            content_bytes = client.download_blob().content_as_bytes()
 
             try:
-                new.value = new.serializer.deserialize(content_string)
+                new.value = new.serializer.deserialize(content_bytes)
             except EOFError:
                 new.value = None
             self.logger.debug("Finished downloading result from {}.".format(location))


Test patch:
diff --git a/tests/engine/cloud/test_cloud_task_runner.py b/tests/engine/cloud/test_cloud_task_runner.py
index fb8d244f3e..5a0a833e34 100644
--- a/tests/engine/cloud/test_cloud_task_runner.py
+++ b/tests/engine/cloud/test_cloud_task_runner.py
@@ -480,6 +480,62 @@ def test_task_runner_prioritizes_kwarg_states_over_db_states(monkeypatch, state)
     assert [type(s).__name__ for s in states] == ["Running", "Success"]
 
 
+class TestHeartBeats:
+    def test_heartbeat_traps_errors_caused_by_client(self, caplog, monkeypatch):
+        client = MagicMock(graphql=MagicMock(side_effect=SyntaxError))
+        monkeypatch.setattr(
+            "prefect.engine.cloud.task_runner.Client", MagicMock(return_value=client)
+        )
+        runner = CloudTaskRunner(task=Task(name="bad"))
+        runner.task_run_id = None
+        res = runner._heartbeat()
+        assert res is False
+
+        log = caplog.records[0]
+        assert log.levelname == "ERROR"
+        assert "Heartbeat failed for Task 'bad'" in log.message
+
+    def test_heartbeat_traps_errors_caused_by_bad_attributes(self, caplog, monkeypatch):
+        monkeypatch.setattr("prefect.engine.cloud.task_runner.Client", MagicMock())
+        runner = CloudTaskRunner(task=Task())
+        res = runner._heartbeat()
+        assert res is False
+
+        log = caplog.records[0]
+        assert log.levelname == "ERROR"
+        assert "Heartbeat failed for Task 'Task'" in log.message
+
+    @pytest.mark.parametrize("setting_available", [True, False])
+    def test_task_runner_heartbeat_sets_command(self, monkeypatch, setting_available):
+        client = MagicMock()
+        monkeypatch.setattr(
+            "prefect.engine.cloud.task_runner.Client", MagicMock(return_value=client)
+        )
+        client.graphql.return_value.data.flow_run_by_pk.flow.settings = (
+            dict(heartbeat_enabled=True) if setting_available else {}
+        )
+
+        runner = CloudTaskRunner(task=Task())
+        runner.task_run_id = "foo"
+        res = runner._heartbeat()
+        assert res is True
+        assert runner.task_run_id == "foo"
+        assert runner.heartbeat_cmd == ["prefect", "heartbeat", "task-run", "-i", "foo"]
+
+    def test_task_runner_does_not_have_heartbeat_if_disabled(self, monkeypatch):
+        client = MagicMock()
+        monkeypatch.setattr(
+            "prefect.engine.cloud.task_runner.Client", MagicMock(return_value=client)
+        )
+        client.graphql.return_value.data.flow_run_by_pk.flow.settings = dict(
+            heartbeat_enabled=False
+        )
+        runner = CloudTaskRunner(task=Task())
+        runner.task_run_id = "foo"
+        res = runner._heartbeat()
+        assert res is False
+
+
 class TestStateResultHandling:
     def test_task_runner_handles_outputs_prior_to_setting_state(self, client):
         @prefect.task(cache_for=datetime.timedelta(days=1), result=PrefectResult())
diff --git a/tests/engine/results/test_azure_result.py b/tests/engine/results/test_azure_result.py
index cca20be546..cddcd7e84e 100644
--- a/tests/engine/results/test_azure_result.py
+++ b/tests/engine/results/test_azure_result.py
@@ -58,7 +58,11 @@ class TestAzureResult:
         assert service.get_blob_client.call_args[1]["container"] == "foo"
 
     def test_azure_reads_and_updates_location(self, monkeypatch):
-        client = MagicMock(download_blob=MagicMock(return_value=b""))
+        client = MagicMock(
+            download_blob=MagicMock(
+                return_value=MagicMock(content_as_bytes=MagicMock(return_value=b""))
+            )
+        )
         service = MagicMock(get_blob_client=MagicMock(return_value=client))
         monkeypatch.setattr(
             "prefect.engine.results.azure_result.AzureResult.service", service