1115.bugfix: Make KafkaStorageError retriable after metadata refresh (#1115)

* Make KafkaStorageError retriable after metadata refresh

This is how it is in the java implementation and in kafka-python

[Java Implementation](https://kafka.apache.org/30/javadoc/org/apache/kafka/common/errors/KafkaStorageException.html)
```
Client should request metadata update and retry if the response shows KafkaStorageException
```

[kafka-python](https://github.com/dpkp/kafka-python/blob/20e2d52ff2c337e02f8bac7af6c6e8d00ebcc63f/kafka/errors.py#L546)
```Python
class KafkaStorageError(BrokerResponseError):
    errno = 56
    message = 'KAFKA_STORAGE_ERROR'
    description = 'Disk error when trying to access log file on the disk.'
    retriable = True
    invalid_metadata = True
```

* Update CHANGES.rst

* Fix formatting

---------

Co-authored-by: Denis Otkidach <[email protected]>

Author: omerhadari

CHANGES.rst

@@ -11,6 +11,13 @@ Improved Documentation:
   (pr #1068 by @jzvandenoever)
 * Fix Java Client API reference (pr #1069 by @emmanuel-ferdman)
 
+
+Bugfixes:
+
+* Make KafkaStorageError retriable after metadata refresh like in other
+  implementations (pr #1115 by @omerhadari)
+
+
 Misc:
 * Use SPDX license expression for project metadata.
 

aiokafka/errors.py

@@ -740,7 +740,9 @@ class OperationNotAttempted(BrokerResponseError):
 class KafkaStorageError(BrokerResponseError):
     errno = 56
     message = "KAFKA_STORAGE_ERROR"
-    description = "The user-specified log directory is not found in the broker config."
+    description = "Disk error when trying to access log file on the disk."
+    retriable = True
+    invalid_metadata = True
 
 
 class LogDirNotFound(BrokerResponseError):