add details to message reading logs (#1943)

* add details to schema conversion logs

* lint

* lint

Author: moscicky

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/receiver/kafka/BasicMessageContentReader.java

@@ -53,7 +53,11 @@ private void ensureExistence(Integer schemaVersion, Integer schemaId) {
         schemaExistenceEnsurer.ensureSchemaExists(topic, SchemaId.valueOf(schemaId));
       }
     } catch (SchemaExistenceEnsurer.SchemaNotLoaded ex) {
-      throw new RetryableReceiverError("Requested schema not present yet...", ex);
+      throw new RetryableReceiverError(
+          String.format(
+              "Requested schema not present yet for topic: %s, schemaVersion: %s, schemaId: %s",
+              topic.getQualifiedName(), schemaVersion, schemaId),
+          ex);
     }
   }
 }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/receiver/kafka/KafkaSingleThreadedMessageReceiver.java

@@ -148,7 +148,10 @@ private Optional<Message> getMessageFromReadQueue() {
         readQueue.poll();
         return Optional.of(message);
       } catch (RetryableReceiverError ex) {
-        logger.warn("Cannot convert record to message... Operation will be delayed", ex);
+        logger.warn(
+            "Cannot convert record to message for subscription {}, operation will be delayed",
+            subscription.getQualifiedName(),
+            ex);
         return Optional.empty();
       }
     }