fix(ingest/kafka): update warning reporting for kafka (#11171)

metadata-ingestion/src/datahub/ingestion/source/confluent_schema_registry.py

@@ -249,31 +249,33 @@ def _get_schema_and_fields(
 
         if topic_subject is not None:
             logger.debug(
-                f"The {schema_type_str} schema subject:'{topic_subject}' is found for {kafka_entity}:'{topic}'."
+                f"The {schema_type_str} schema subject:'{topic_subject}' is found for {kafka_entity}: '{topic}'."
             )
             try:
                 registered_schema = self.schema_registry_client.get_latest_version(
                     subject_name=topic_subject
                 )
                 schema = registered_schema.schema
             except Exception as e:
-                logger.warning(
-                    f"For {kafka_entity}: {topic}, failed to get {schema_type_str} schema from schema registry using subject:'{topic_subject}': {e}."
-                )
-                self.report.report_warning(
-                    topic,
-                    f"failed to get {schema_type_str} schema from schema registry using subject:'{topic_subject}': {e}.",
+                self.report.warning(
+                    title="Failed to get subject schema from schema registry",
+                    message=f"Failed to get {kafka_entity} {schema_type_str or ''} schema from schema registry",
+                    context=f"{topic}: {topic_subject}"
+                    if not is_subject
+                    else topic_subject,
+                    exc=e,
                 )
         else:
             logger.debug(
                 f"For {kafka_entity}: {topic}, the schema registry subject for the {schema_type_str} schema is not found."
             )
             if not is_key_schema:
                 # Value schema is always expected. Report a warning.
-                self.report.report_warning(
-                    topic,
-                    f"The schema registry subject for the {schema_type_str} schema is not found."
-                    f" The {kafka_entity} is either schema-less, or no messages have been written to the {kafka_entity} yet.",
+                self.report.warning(
+                    title="Unable to find a matching subject name for the topic in the schema registry",
+                    message=f"The {kafka_entity} {schema_type_str or ''} is either schema-less, or no messages have been written to the {kafka_entity} yet. "
+                    "If this is unexpected, check the topic_subject_map and topic_naming related configs.",
+                    context=topic,
                 )
 
         # Obtain the schema fields from schema for the topic.