Kafka/long lived producer

[guillermodotn]

Addresses sections 3 and 4 commented on PR #82

3.- There is batching of messages. The producer does this automatically (partially).
4.- The producer is created for each batch of messages. Kafka seems to prefer a long lived producer that is reused.

Comment ref: #82 (review)
Stacked on: #82

[codecov-commenter]

Codecov Report

❌ Patch coverage is 90.90909% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.76%. Comparing base (34cc0d8) to head (d4e985a).

Files with missing lines	Patch %	Lines
cts/messaging.py	90.90%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main      #83      +/-   ##
==========================================
+ Coverage   83.65%   83.76%   +0.10%     
==========================================
  Files          13       13              
  Lines        1328     1337       +9     
==========================================
+ Hits         1111     1120       +9     
  Misses        217      217              

Flag	Coverage Δ
unit-tests	83.76% <90.90%> (+0.10%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[lubomir]

I like the general direction. However, I think we should wait with further changes until we confirm in stage that the code is indeed working so that we don't have to chase a moving target.

[guillermodotn]

Kafka long-lived producer, kafka-python handles send retries; flush() for delivery confirmation/logging and on failure recreate the producer. _retry_with_backoff unchanged.

[lubomir]

There is a race condition risk with this global mutable variable. The main thread could be closing the producer at the same time as the executor thread is trying to publish a message. But the closing happens only at exit, so the race is not really a big deal as the application is shutting down anyway. Once UMB is removed, the threadpool will go to and this problem will mostly go away. Doesn't seem like a huge deal.