kumarpritam863 commented on issue #16282:
URL: https://github.com/apache/iceberg/issues/16282#issuecomment-5099961901
@tongwai-wong-appier
Agreed, but what I was thinking is shouldn't this change sufficient along
with open/close and zombie coordinator gurad changes that is there in the PR:
```
protected void consumeAvailable(Duration pollDuration) {
ConsumerRecords<String, byte[]> records = consumer.poll(pollDuration);
while (!records.isEmpty()) {
for (ConsumerRecord<String, byte[]> record : records) {
// the consumer stores the offset of the next record to consume,
// so increment the record offset by one
if (record.offset() <
controlTopicOffsets.getOrDefault(record.partition(), 0L)) {
LOG.warn("Channel {} processed an already processed offset {} for
partition {}", taskId, record.offset(), record.partition());
continue;
}
controlTopicOffsets.put(record.partition(), record.offset() + 1);
Event event = AvroUtil.decode(record.value());
if (event.groupId().equals(connectGroupId)) {
LOG.debug("Received event of type: {}", event.type().name());
if (receive(new Envelope(event, record.partition(),
record.offset()))) {
LOG.info("Handled event of type: {}", event.type().name());
}
}
}
records = consumer.poll(pollDuration);
}
}
```
As neither the worker not the coordinator would want to read a record which
is already read by them in any of the scenarios.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
