Mqtt-kafka single group support cont #731
Conversation
There was a problem hiding this comment.
We also need scenario(s) to check the behavior when producer id changes on the same client produce stream, either to a different producer id value or to no producer id.
This will happen frequently if multiple clients are producing at a mixture of qos to the same kafka topic such as mqtt-messages.
| @@ -542,11 +545,21 @@ private int flushRecordInit( | |||
| final int maxEncodeableBytes = client.encodeSlotLimit + client.valueCompleteSize + produceRecordFramingSize; | |||
|
|
|||
| if (client.encodeSlot != NO_SLOT && | |||
| maxEncodeableBytes > encodePool.slotCapacity()) | |||
| (maxEncodeableBytes > encodePool.slotCapacity() || | |||
| client.producerId != producerId && client.producerEpoch != producerEpoch)) | |||
There was a problem hiding this comment.
| client.producerId != producerId && client.producerEpoch != producerEpoch)) | |
| client.producerId != producerId || client.producerEpoch != producerEpoch)) |
only need one of these to differ to trigger request since both are expected to be consistent for entire batch, right?
| .baseSequence(RECORD_BATCH_SEQUENCE_NONE) | ||
| .producerId(client.producerId) | ||
| .producerEpoch(client.producerEpoch) | ||
| .baseSequence(sequence) | ||
| .recordCount(encodeableRecordCount) | ||
| .build(); |
There was a problem hiding this comment.
Need to reset producerId and producerEpoch after encoding request so that next request can batch as expected?
| { | ||
| client.doEncodeRequestIfNecessary(traceId, budgetId); | ||
| } | ||
|
|
||
| if (client.producerId == RECORD_BATCH_PRODUCER_ID_NONE || | ||
| client.producerId != producerId) |
There was a problem hiding this comment.
Should this be && instead?
Please remove extra whitespace around !=.
| if (client.producerId == RECORD_BATCH_PRODUCER_ID_NONE || | ||
| client.producerId != producerId) | ||
| { | ||
| client.sequence = sequence; |
There was a problem hiding this comment.
Suggest we rename to client.baseSequence.
| .typeId(zilla:id("kafka")) | ||
| .produce() | ||
| .topic("test") | ||
| .partition(1) |
There was a problem hiding this comment.
We need to match producerId and producerEpoch here too, right?
There was a problem hiding this comment.
it is part of dataEx I have filled missing ones. We can't send producerId and producerEpoch as part of beginEx since we don't know the partitions and we need partitions to fetch.
| .typeId(zilla:id("kafka")) | ||
| .produce() | ||
| .topic("test") | ||
| .partition(0) |
There was a problem hiding this comment.
producerId and producerEpoch?
| .merged() | ||
| .produce() | ||
| .key("a") | ||
| .hashKey("key7") |
There was a problem hiding this comment.
Does produce flush need both key and hashKey or just hashKey?
Description
Mqtt-kafka single group support cont
Fixes #698