convert producer message to json

src/main.py

@@ -101,7 +101,7 @@ async def fan_out_msg(
     data
 ):
     logging.info(f"sending msg {data}")
-    await producer.send_and_wait(fan_out_topic, data)
+    await producer.send_and_wait(fan_out_topic, json.dumps(data))
 
 
 async def main() -> None:
@@ -227,6 +227,9 @@ async def main() -> None:
                         logging.info("Message does not have an instrument. Assuming "
                                      "it's not an observation.")
                         continue
+
+                    '''
+                    # Temporary disable so we can see older messages for testing.
 
                     # efdStamp is visit publication, in seconds since 1970-01-01 UTC
                     if next_visit_message_initial["message"]["private_efdStamp"]:
@@ -240,7 +243,7 @@ async def main() -> None:
                             continue
                     else:
                         logging.warning("Message does not have private_efdStamp, can't determine age.")
-
+                    '''
                     next_visit_message_updated = NextVisitModel(
                         salIndex=next_visit_message_initial["message"]["salIndex"],
                         scriptSalIndex=next_visit_message_initial["message"][
@@ -369,8 +372,6 @@ async def main() -> None:
                         await producer.start()
                         logging.info ("started kafka producer")
 
-                        #for fan_out_message in fan_out_message_list:
-
                         for fan_out_message in fan_out_message_list:
 
                             task = asyncio.create_task(