[GitHub] [kafka] hachikuji commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
hachikuji commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r583219431
##
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
##
@@ -475,11 +476,15 @@ public boolean commitOffsets() {
synchronized (this) {
// First we need to make sure we snapshot everything in exactly
the current state. This
// means both the current set of messages we're still waiting to
finish, stored in this
-// class, which setting flushing = true will handle by storing any
new values into a new
+// class, which setting recordFlushPending = true will handle by
storing any new values into a new
// buffer; and the current set of user-specified offsets, stored
in the
// OffsetStorageWriter, for which we can use beginFlush() to
initiate the snapshot.
-flushing = true;
-boolean flushStarted = offsetWriter.beginFlush();
+// No need to begin a new offset flush if we timed out waiting for
records to be flushed to
+// Kafka in a prior attempt.
+if (!recordFlushPending) {
Review comment:
It's mostly the flushing that concerns me, not really the offset commit.
I don't think we need to make it synchronous, just that it seems silly to block
that shared scheduler to complete it. My thought instead was to let the
scheduler trigger the flush, but then let the task be responsible for waiting
for its completion. While waiting, of course, it can continue writing to
`outstandingMessagesBacklog`. So I don't think there should be any issue from a
throughput perspective.
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.
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] hachikuji commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
hachikuji commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r582378783
##
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
##
@@ -475,11 +476,15 @@ public boolean commitOffsets() {
synchronized (this) {
// First we need to make sure we snapshot everything in exactly
the current state. This
// means both the current set of messages we're still waiting to
finish, stored in this
-// class, which setting flushing = true will handle by storing any
new values into a new
+// class, which setting recordFlushPending = true will handle by
storing any new values into a new
// buffer; and the current set of user-specified offsets, stored
in the
// OffsetStorageWriter, for which we can use beginFlush() to
initiate the snapshot.
-flushing = true;
-boolean flushStarted = offsetWriter.beginFlush();
+// No need to begin a new offset flush if we timed out waiting for
records to be flushed to
+// Kafka in a prior attempt.
+if (!recordFlushPending) {
Review comment:
> (Please correct me if I'm wrong on this point; my core knowledge is a
little fuzzy and maybe there are stronger guarantees than I'm aware of)
Out-of-order acknowledgment of records makes tracking the latest offset for a
given source partition a little less trivial than it seems initially. For
example, if a task produces two records with the same source partition that end
up being delivered to different topic-partitions, the second record may be
ack'd before the first, and when it comes time for offset commit, the framework
would have to refrain from committing offsets for that second record until the
first is also ack'd.
Ok, that rings a bell. I think I see how the logic works now and I don't see
an obvious way to make it simpler. Doing something finer-grained as you said
might be the way to go. Anyway, I agree this is something to save for a
follow-up improvement.
> I think it's a necessary evil, since source task offset commits are
conducted on a single thread. Without a timeout for offset commits, a single
task could block indefinitely and disable offset commits for all other tasks on
the cluster.
Hmm.. This is suspicious. Why do we need to block the executor while we wait
for the flush? Would it be simpler to let the worker source task finish the
flush and the offset commit in its own event thread? We end up blocking the
event thread anyway because of the need to do it under the lock.
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.
For queries about this service, please contact Infrastructure at:
[email protected]
[GitHub] [kafka] hachikuji commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
hachikuji commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r582259574
##
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
##
@@ -98,7 +98,8 @@
private IdentityHashMap,
ProducerRecord> outstandingMessages;
// A second buffer is used while an offset flush is running
private IdentityHashMap,
ProducerRecord> outstandingMessagesBacklog;
-private boolean flushing;
+private boolean recordFlushPending;
+private boolean offsetFlushPending;
private CountDownLatch stopRequestedLatch;
Review comment:
nit: while we're at it, this could be `final`
##
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
##
@@ -475,11 +476,15 @@ public boolean commitOffsets() {
synchronized (this) {
// First we need to make sure we snapshot everything in exactly
the current state. This
// means both the current set of messages we're still waiting to
finish, stored in this
-// class, which setting flushing = true will handle by storing any
new values into a new
+// class, which setting recordFlushPending = true will handle by
storing any new values into a new
// buffer; and the current set of user-specified offsets, stored
in the
// OffsetStorageWriter, for which we can use beginFlush() to
initiate the snapshot.
-flushing = true;
-boolean flushStarted = offsetWriter.beginFlush();
+// No need to begin a new offset flush if we timed out waiting for
records to be flushed to
+// Kafka in a prior attempt.
+if (!recordFlushPending) {
Review comment:
If I understand it correctly, the main difference in this patch is that
we no longer fail the flush if the messages cannot be drained quickly enough
from `outstandingMessages`. A few questions come to mind:
1. Is the flush timeout still a useful configuration? Was it ever? Even if
we timeout, we still have to wait for the records that were sent to the
producer.
2. While we are waiting for `outstandingMessages` to be drained, we are
still accumulating messages in `outstandingMessagesBacklog`. I imagine we can
get into a pattern here once we fill up the accumulator. While we're waiting
for `outstandingMessages` to complete, we fill `outstandingMessagesBacklog`.
Once the flush completes, `outstandingMessagesBacklog` becomes
`outstandingMessages` and we are stuck waiting again. Could this prevent us
from satisfying the commit interval?
Overall, I can't shake the feeling that this logic is more complicated than
necessary. Why do we need the concept of flushing at all? It would be more
intuitive to just commit whatever the latest offsets are. Note that we do not
use `outstandingMessages` for the purpose of retries. Once a request has been
handed off to the producer successfully, we rely on the producer to handle
retries. Any delivery failure after that is treated as fatal. So then does
`oustandingMessages` serve any other purpose other than tracking flushing? I am
probably missing something here. It has been a long time since I reviewed this
logic.
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.
For queries about this service, please contact Infrastructure at:
[email protected]
