[GitHub] [kafka] C0urante commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
C0urante commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r589603947
##
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:
I've been ruminating over this for a few days and I think it should be
possible to make task offset commits independent of each other by changing the
source task offset commit scheduler to use a multi-threaded executor instead of
a global single-threaded executor for all tasks. This isn't quite the same
thing as what you're proposing since tasks would still not be responsible for
waiting for flush completion (the offset scheduler's threads would be), but
it's a smaller change and as far as I can tell, the potential downsides only
really amount to a few extra threads being created.
The usage of
[`scheduleWithFixedDelay`](https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ScheduledExecutorService.html#scheduleWithFixedDelay-java.lang.Runnable-long-long-java.util.concurrent.TimeUnit-)
already ensures that two offset commits for the same task won't be active at
the same time, as it "Creates and executes a periodic action that becomes
enabled first after the given initial delay, and subsequently with the given
delay between the termination of one execution and the commencement of the
next."
Beyond that, the only concern that comes to mind is potential races caused
by concurrent access of the offset backing store and its underlying resources.
In distributed mode, the `KafkaOffsetBackingStore` and its usage of the
underlying `KafkaBasedLog` appear to be thread-safe as everything basically
boils down to calls to `Producer::send`, which should be fine.
In standalone mode, the `MemoryOffsetBackingStore` handles all writes/reads
of the local offsets file via a single-threaded executor, so concurrent calls
to `MemoryOffsetBackingStore::set` should also be fine.
Granted, none of this addresses your original concern, which is whether an
offset commit timeout is necessary at all. In response to that, I think we may
also want to revisit the offset commit logic and possibly do away with a
timeout altogether. In sink tasks, for example, offset commit timeouts are
almost a cosmetic feature at this point and are really only useful for metrics
tracking. However, at the moment it's actually been pretty useful to us to
monitor source task offset commit success/failure JMX metrics as a means of
tracking overall task health. We might be able to make up the difference by
relying on metrics for the number of active records, but it's probably not safe
to make that assumption for all users, especially for what is intended to be a
bug fix. So, if possible, I'd like to leave a lot of the offset commit logic
intact as it is for the moment and try to keep the changes here minimal.
To summarize: I'd like to proceed by keeping the currently-proposed changes,
and changing the source task offset committer to use a multi-threaded executor
instead of a single-threaded executor. I can file a follow-up ticket to track
improvements in offset commit logic (definitely for source tasks, and possibly
for sinks) and we can look into that if it becomes a problem in the future.
What do you think?
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] C0urante commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
C0urante commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r583154155
##
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:
> We end up blocking the event thread anyway because of the need to do
it under the lock.
I think we actually keep polling the task for records during the offset
commit, which is the entire reason we have the `outstandingMessagesBacklog`
field. Without it, we'd just add everything to `outstandingMessages` knowing
that, if we've made it to the point of adding a record to that collection,
we're not in the process of committing offset, right?
Concretely, we can see that the offset thread [relinquishes the lock on the
`WorkerSourceTask` instance while waiting for outstanding messages to be
ack'd](https://github.com/C0urante/kafka/blob/03c5a83a8277fa7c4ec503c3e044ae61cff06eea/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java#L501).
I'm not sure we _need_ to perform offset commits on a separate thread, but
it is in line with what we do for sink tasks, where we [leverage the
`Consumer::commitAsync`
method](https://github.com/apache/kafka/blob/e2a0d0c90e1916d77223a420e3595e8aba643001/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L365).
If we want to consider making offset commit synchronous (which is likely
going to happen anyways when transactional writes for exactly-once source are
introduced), that also might be worth a follow-up. The biggest problem I can
think of with that approach would be that a single offline topic-partition
would block up the entire task thread when it comes time for offset commit. If
we keep the timeout for offset commit, then that'd limit the fallout and allow
us to resume polling new records from the task and dispatching them to the
producer after the commit attempt timed out. However, there'd still be a
non-negligible throughput hit (especially for workers configured with higher
offset timeouts).
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] C0urante commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
C0urante commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r583154155
##
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:
> We end up blocking the event thread anyway because of the need to do
it under the lock.
I think we actually keep polling the task for records during the offset
commit, which is the entire reason we have the `outstandingMessagesBacklog`
field. Without it, we'd just add everything to `outstandingMessages` knowing
that, if we've made it to the point of adding a record to that collection,
we're not in the process of committing offset, right?
Concretely, we can see that the offset thread [relinquishes the lock on the
`WorkerSourceTask` instance while waiting for outstanding messages to be
ack'd](https://github.com/C0urante/kafka/blob/03c5a83a8277fa7c4ec503c3e044ae61cff06eea/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java).
I'm not sure we _need_ to perform offset commits on a separate thread, but
it is in line with what we do for sink tasks, where we [leverage the
`Consumer::commitAsync`
method](https://github.com/apache/kafka/blob/e2a0d0c90e1916d77223a420e3595e8aba643001/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java#L365).
If we want to consider making offset commit synchronous (which is likely
going to happen anyways when transactional writes for exactly-once source are
introduced), that also might be worth a follow-up. The biggest problem I can
think of with that approach would be that a single offline topic-partition
would block up the entire task thread when it comes time for offset commit. If
we keep the timeout for offset commit, then that'd limit the fallout and allow
us to resume polling new records from the task and dispatching them to the
producer after the commit attempt timed out. However, there'd still be a
non-negligible throughput hit (especially for workers configured with higher
offset timeouts).
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] C0urante commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
C0urante commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r582351074
##
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:
1. I think it's a necessary evil, since source task offset commits are
[conducted on a single
thread](https://github.com/apache/kafka/blob/3f09fb97b6943c0612488dfa8e5eab8078fd7ca0/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java#L64).
Without a timeout for offset commits, a single task could block indefinitely
and disable offset commits for all other tasks on the cluster.
2. This is definitely possible; I think the only saving grace here is that
the combined sizes of the `outstandingMessages` and
`outstandingMessagesBacklog` fields is going to be naturally throttled by the
producer's buffer. If too many records are accumulated, the call to
`Producer::send` will block synchronously until space is freed up, at which
point, the worker can continue polling the task for new records. This isn't
ideal as it will essentially cause the producer's entire buffer to be occupied
until the throughput of record production from the task decreases and/or the
write throughput of the producer rises to meet it, but it at least establishes
an upper bound for how large a single batch of records in the
`oustandingMessages` field ever gets. It may take several offset commit
attempts for all of the records in that batch to be ack'd, with all but the
last (successful) attempt timing out and failing, but forward progress with
offset commits should still be possible.
I share your feelings about the complexity here. I think ultimately it
arises from two constraints:
1. A worker-global producer is used to write source offsets to the internal
offsets topic right now. Although this doesn't necessarily require the
single-threaded logic for offset commits mentioned above, things become simpler
with it.
2. (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.
I don't think either of these points make it impossible to add even
more-fine-grained offset commit behavior and/or remove offset commit timeouts,
but the work involved would be a fair amount heavier than this relatively-minor
patch. If you'd prefer to see something along those lines, could we consider
merging this patch for the moment and perform a more serious overhaul of the
source task offset commit logic as a follow-up, possibly with a small design
discussion on a Jira ticket to make sure there's alignment on the new behavior?
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] C0urante commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
C0urante commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r582351074
##
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:
1. I think it's a necessary evil, since source task offset commits are
[conducted on a single
thread](https://github.com/apache/kafka/blob/3f09fb97b6943c0612488dfa8e5eab8078fd7ca0/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java#L64).
Without a timeout for offset commits, a single task could block indefinitely
and disable offset commits for all other tasks on the cluster.
2. This is definitely possible; I think the only saving grace here is that
the combined sizes of the `outstandingMessages` and
`outstandingMessagesBacklog` fields is going to be naturally throttled by the
producer's buffer. If too many records are accumulated, the call to
`Producer::send` will block synchronously until space is freed up, at which
point, the worker can continue polling the task for new records. This isn't
ideal as it will essentially cause the producer's entire buffer to be occupied
until the throughput of record production from the task decreases and/or the
write throughput of the producer rises to meet it, but it at least establishes
an upper bound for how large a single batch of records in the
`oustandingMessages` field ever gets. It may take several offset commit
attempts for all of the records in that batch to be ack'd, with all but the
last (successful) attempt timing out and failing, but forward progress with
offset commits should still be possible.
I share your feelings about the complexity here. I think ultimately it
arises from two constraints:
1. A worker-global producer is used to write source offsets to the internal
offsets topic right now. Although this doesn't necessarily require the
single-threaded logic for offset commits mentioned above, things become simpler
with it.
2. (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.
I don't think either of these points make it impossible to add even
more-fine-grained offset commit behavior and/or remove offset commit timeouts,
but the work involved would be a fair amount heavier than this relatively-minor
patch. If you'd prefer to see something along those lines, could we merge this
patch for the moment and perform a more serious overhaul of the source task
offset commit logic as a follow-up, possibly with a small design discussion on
a Jira ticket to make sure there's alignment on the new behavior?
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] C0urante commented on a change in pull request #10112: KAFKA-12226: Prevent source task offset failure when producer is overwhelmed
C0urante commented on a change in pull request #10112:
URL: https://github.com/apache/kafka/pull/10112#discussion_r578661623
##
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java
##
@@ -1100,7 +1233,7 @@ private void assertPollMetrics(int
minimumPollCountExpected) {
double activeCountMax =
metrics.currentMetricValueAsDouble(sourceTaskGroup,
"source-record-active-count-max");
assertEquals(0, activeCount, 0.01d);
if (minimumPollCountExpected > 0) {
-assertEquals(RECORDS.size(), activeCountMax, 0.01d);
+assertEquals(activeCountMaxExpected, activeCountMaxExpected,
0.01d);
Review comment:
Ah yeah, good catch. Thanks!
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]
