temp: sad path poc

edx_event_bus_kafka/internal/exceptions.py

@@ -0,0 +1,6 @@
+class BadConfigurationException(Exception):
+    pass
+
+
+class MissingKeyException(Exception):
+    pass

edx_event_bus_kafka/internal/producer.py

@@ -19,6 +19,7 @@
 from openedx_events.tooling import OpenEdxPublicSignal
 
 from .config import get_schema_registry_client, load_common_settings
+from .exceptions import BadConfigurationException, MissingKeyException
 
 logger = logging.getLogger(__name__)
 
@@ -27,14 +28,16 @@
     import confluent_kafka
     from confluent_kafka import Producer
     from confluent_kafka.schema_registry.avro import AvroSerializer
-    from confluent_kafka.serialization import MessageField, SerializationContext
+    from confluent_kafka.serialization import MessageField, SerializationContext, StringSerializer
 except ImportError:  # pragma: no cover
     confluent_kafka = None
 
 # CloudEvent standard name for the event type header, see
 # https://github.com/cloudevents/spec/blob/v1.0.1/kafka-protocol-binding.md#325-example
 EVENT_TYPE_HEADER_KEY = "ce_type"
 
+LOG_FORMAT_STRING = "Error sending message. Message: {} Error: {}"
+
 
 def extract_event_key(event_data: dict, event_key_field: str) -> Any:
     """
@@ -53,14 +56,14 @@ def extract_event_key(event_data: dict, event_key_field: str) -> Any:
     for field_name in field_path:
         if isinstance(current_data, dict):
             if field_name not in current_data:
-                raise Exception(
+                raise MissingKeyException(
                     f"Could not extract key from event; lookup in {event_key_field} "
                     f"failed at {field_name!r} in dictionary"
                 )
             current_data = current_data[field_name]
         else:
             if not hasattr(current_data, field_name):
-                raise Exception(
+                raise MissingKeyException(
                     f"Could not extract key from event; lookup in {event_key_field} "
                     f"failed at {field_name!r} in object"
                 )
@@ -134,7 +137,7 @@ def get_serializers(signal: OpenEdxPublicSignal, event_key_field: str):
     """
     client = get_schema_registry_client()
     if client is None:
-        raise Exception('Cannot create Kafka serializers -- missing library or settings')
+        raise BadConfigurationException('Cannot create Kafka serializers -- missing library or settings')
 
     signal_serializer = AvroSignalSerializer(signal)
 
@@ -191,16 +194,23 @@ def send(
               string naming the dictionary keys to descend)
             event_data: The event data (kwargs) sent to the signal
         """
-        event_key = extract_event_key(event_data, event_key_field)
-        headers = {EVENT_TYPE_HEADER_KEY: signal.event_type}
-
-        key_serializer, value_serializer = get_serializers(signal, event_key_field)
-        key_bytes = key_serializer(event_key, SerializationContext(topic, MessageField.KEY, headers))
-        value_bytes = value_serializer(event_data, SerializationContext(topic, MessageField.VALUE, headers))
-
-        self.producer.produce(
-            topic, key=key_bytes, value=value_bytes, headers=headers, on_delivery=on_event_deliver,
-        )
+        try:
+            event_key = extract_event_key(event_data, event_key_field)
+            headers = {EVENT_TYPE_HEADER_KEY: signal.event_type}
+
+            key_serializer, value_serializer = get_serializers(signal, event_key_field)
+            key_bytes = key_serializer(event_key, SerializationContext(topic, MessageField.KEY, headers))
+            value_bytes = value_serializer(event_data, SerializationContext(topic, MessageField.VALUE, headers))
+
+            self.producer.produce(
+                topic, key=key_bytes, value=value_bytes, headers=headers, on_delivery=on_event_deliver,
+            )
+        except BadConfigurationException as bce:
+            logger.error(LOG_FORMAT_STRING.format(str(event_data), bce))
+        except MissingKeyException as mke:
+            self._send_to_error_topic(topic=topic, event_data=event_data)
+        except:
+            self._send_to_error_topic(topic=topic, event_data=event_data, event_key=extract_event_key(event_data, event_key_field))
 
         # Opportunistically ensure any pending callbacks from recent event-sends are triggered.
         # This ensures that we're polling at least as often as we're producing, which is a
@@ -210,6 +220,16 @@ def send(
         # poll(0) on a regular interval (see `poll_indefinitely`).
         self.producer.poll(0)
 
+    def _send_to_error_topic(self, *, topic: str, event_data: dict, event_key = None):
+        try:
+            event_key_as_str = str(event_key if event_key else "Missing Key")
+            event_data_as_str = json.dumps(event_data, allow_nan=True, skipkeys=True, default=lambda x: str(x))
+            self.producer.produce(f"{topic}-error", key=event_key_as_str, value=event_data_as_str)
+        except e:
+            logger.error(str(event_data), e)
+
+
+
     def prepare_for_shutdown(self):
         """
         Prepare producer for a clean shutdown.
@@ -280,8 +300,9 @@ def get_producer() -> Optional[EventProducerKafka]:
     producer_settings = load_common_settings()
     if producer_settings is None:
         return None
+    thing = Producer(producer_settings)
 
-    return EventProducerKafka(Producer(producer_settings))
+    return EventProducerKafka(thing)
 
 
 def on_event_deliver(err, evt):
@@ -297,6 +318,7 @@ def on_event_deliver(err, evt):
     """
     if err is not None:
         logger.warning(f"Event delivery failed: {err!r}")
+        logger.error(LOG_FORMAT_STRING.format(evt, err))
     else:
         # Don't log msg.value() because it may contain userids and/or emails
         logger.info(f"Event delivered to topic {evt.topic()}; key={evt.key()}; "
@@ -306,5 +328,6 @@ def on_event_deliver(err, evt):
 @receiver(setting_changed)
 def _reset_caches(sender, **kwargs):  # pylint: disable=unused-argument
     """Reset caches when settings change during unit tests."""
+    logger.info("Clearin'")
     get_serializers.cache_clear()
     get_producer.cache_clear()