add optional kafka stream as a secondary endpoint to caduceus #562
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #562 +/- ##
==========================================
+ Coverage 58.94% 62.20% +3.26%
==========================================
Files 18 19 +1
Lines 1510 1717 +207
==========================================
+ Hits 890 1068 +178
- Misses 577 602 +25
- Partials 43 47 +4
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
eventDispatcher.go
Outdated
|
|
||
| // OnDeviceEvent is the device.Listener function that processes outbound events. | ||
| func (d *eventDispatcher) OnDeviceEvent(event *device.Event) { | ||
| d.logger.Debug("Received device event", zap.Any("event", event)) |
There was a problem hiding this comment.
What is the performance/memory impact of this call? This is a very high throughput call & this may be totally fine or it may cause memory pressure.
There was a problem hiding this comment.
it will have an impact because of reflection so it's removed
…me any messsages from kafka broker
publisher.go
Outdated
| // Convert wrp v3 message to v5 for wrpkafka | ||
| v5msg := convertV3ToV5(msg) | ||
|
|
||
| // *** find better way to do this - set QoS to Critical if in test mode so the message flushes immediately ** |
There was a problem hiding this comment.
Should never change production code for testing. Not sure how else to get the message to flush during an integration test. Setting a linger of 10ms still did not get the message to flush before the test timed out. @schmidtw
There was a problem hiding this comment.
goofy workaround is gone now. works ok now, unclear why it wasn't before. linger is also set back to 0.
see KAFKA_IMPLEMENTION.md for changes. This file can be removed prior to merge.
Fixing AI generated garbage in the integration test.