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 84.61538% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.73%. Comparing base (a24ff29) to head (da3334f).

Files with missing lines	Patch %	Lines
cts/messaging.py	84.61%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main      #83      +/-   ##
==========================================
+ Coverage   83.69%   83.73%   +0.03%     
==========================================
  Files          13       13              
  Lines        1325     1328       +3     
==========================================
+ Hits         1109     1112       +3     
  Misses        216      216              

Flag	Coverage Δ
unit-tests	83.73% <84.61%> (+0.03%)	⬆️

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.

[lubomir]

Does this actually ever raise any exceptions? It returns a Future immediately, so I would not expecte any network issues to appear as exceptions. Adding back the flush() might help.

[guillermodotn]

flush() was dropped to let Kafka handle batching. CTS sends few messages, and linger_ms=0 means they're sent almost immediately anyway.

But you're right, without flush(), send() just returns a Future and delivery errors are never raised. The error recovery code would never trigger. I'll add it back.

[lubomir]

The built-in batching is a good point though. I didn't think about that. Maybe it's the error handling that should be removed?

[lubomir]

This code seems rather fragile. There's a helper to create the producer, but here we still need to touch the global variable directly. Does KafkaProducer have some reconnection logic we could use instead?

[guillermodotn]

Roger, relying on Kafka's built-in reconnection and retry API instead.