[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1214236252
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,30 +280,47 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.entrySet()
+.stream()
+.anyMatch(offset -> offset.getValue() == null);
+
return primaryStore.set(values, (primaryWriteError, ignored) -> {
if (secondaryStore != null) {
if (primaryWriteError != null) {
log.trace("Skipping offsets write to secondary store
because primary write has failed", primaryWriteError);
+try (LoggingContext context = loggingContext()) {
+callback.onCompletion(primaryWriteError, ignored);
+}
} else {
try {
// Invoke OffsetBackingStore::set but ignore the
resulting future; we don't block on writes to this
-// backing store.
+// backing store. The only exception to this is when a
batch consisting of tombstone records fails
+// to be written to secondary store and has been
successfully written to the primary store. In this case
+// an error would be propagated back as in such cases,
a deleted source partition
+// would be reported as present because the 2 stores
are not in sync.
secondaryStore.set(values, (secondaryWriteError,
ignored2) -> {
try (LoggingContext context = loggingContext()) {
-if (secondaryWriteError != null) {
+if (secondaryWriteError != null &&
containsTombstones) {
+log.warn("Failed to write offsets with
tombstone records to secondary backing store", secondaryWriteError);
+callback.onCompletion(secondaryWriteError,
ignored);
+return;
+} else if (secondaryWriteError != null) {
log.warn("Failed to write offsets to
secondary backing store", secondaryWriteError);
} else {
log.debug("Successfully flushed offsets to
secondary backing store");
}
+//primaryWriteError is null at this point, and
we don't care about secondaryWriteError
+callback.onCompletion(null, ignored);
Review Comment:
Isn't this going to cause us to always block on writes to the secondary
store which is something we want to avoid?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1219596953
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,10 +280,33 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
Review Comment:
This `ConnectorOffsetBackingStore::set` method's Javadoc also needs to be
updated to mention the special case handling for batches with `null` offsets
since it currently states the following:
```
* If configured to use a connector-specific offset store, the
returned {@link Future} corresponds to a
* write to that store, and the passed-in {@link Callback} is invoked
once that write completes. If a worker-global
* store is provided, a secondary write is made to that store if the
write to the connector-specific store
* succeeds. Errors with this secondary write are not reflected in the
returned {@link Future} or the passed-in
* {@link Callback}; they are only logged as a warning to users.
```
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,10 +280,33 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.entrySet()
+.stream()
+.anyMatch(offset -> offset.getValue() == null);
Review Comment:
nit: can be simplified
```suggestion
boolean containsTombstones = values.containsValue(null);
```
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,10 +280,33 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.entrySet()
+.stream()
+.anyMatch(offset -> offset.getValue() == null);
+
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+secondaryStore.set(values, (secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteError.compareAndSet(null,
secondaryWriteError);
+} else {
+log.debug("Successfully flushed tombstone offsets to
secondary backing store");
+}
+}
+});
+}
+
return primaryStore.set(values, (primaryWriteError, ignored) -> {
-if (secondaryStore != null) {
+// Secondary store writes have already happened for tombstone
records
Review Comment:
How do we know this if we aren't blocking on the write to the secondary
store above? I believe we should do a synchronous write to the secondary store
in this tombstone offset case.
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -302,7 +326,12 @@ public Future set(Map
values, Callback callb
}
}
try (LoggingContext context = loggingContext()) {
-callback.onCompletion(primaryWriteError, ignored);
+Throwable secondaryWriteError =
secondaryStoreTombstoneWriteError.get();
+if (secondaryStore != null && containsTombstones &&
secondaryWriteError != null) {
Review Comment:
Same as above - we aren't blocking on the write to the secondary store, so
we can't be sure that it has completed at this point.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1219648522
##
connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java:
##
@@ -192,6 +197,220 @@ public void testCancelAfterAwaitFlush() throws Exception {
flushFuture.get(1000, TimeUnit.MILLISECONDS);
}
+@Test
Review Comment:
All these new tests seem to be specific to the `ConnectorOffsetBackingStore`
class - should we move them to a new `ConnectorOffsetBackingStoreTest` class?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1233861544
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -140,6 +142,9 @@ public static ConnectorOffsetBackingStore
withOnlyConnectorStore(
private final Optional connectorStore;
private final Optional connectorStoreAdmin;
+private boolean isEOSEnabled;
Review Comment:
```suggestion
private boolean exactlyOnce;
```
nit: similar to
https://github.com/apache/kafka/blob/9b7f7e0fa09121e204540b46bd3d7340a0175e7d/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java#L152
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,10 +284,61 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+Future secondaryWriteFuture = secondaryStore.set(values,
(secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteError.compareAndSet(null,
secondaryWriteError);
+} else {
+log.debug("Successfully flushed tombstone offsets to
secondary backing store");
+}
+}
+});
Review Comment:
Why do we need to do this if we're anyway doing a get on the returned future
immediately? Any producer error will be wrapped in an `ExecutionException` and
thrown when we call get on the future returned from `KafkaOffsetBackingStore`
(the same exception will also be used to complete the passed callback
exceptionally) -
https://github.com/apache/kafka/blob/546b912b831d46208499c6e1f4f785db6ddd9aa3/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java#L344-L414
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,10 +284,61 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+Future secondaryWriteFuture = secondaryStore.set(values,
(secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteError.compareAndSet(null,
secondaryWriteError);
+} else {
+log.debug("Successfully flushed tombstone offsets to
secondary backing store");
+}
+}
+});
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
+// tombstone records fail to be written to the secondary
store. Note that while commitTransaction
+// already waits for all records to be sent and ack'ed, in
this case we do need to add an explicit
+// blocking call. In case of ALOS, we wait for the same
duration as `offset.commit.timeout.ms`
Review Comment:
nit: I haven't seen the term ALOS used anywhere else in Connect. I'm
presuming it means at least once delivery semantics? The Connect framework
doesn't guarantee at least once semantics for all source connectors - depending
on the specific connector plugin implementation, we could also have at most
onc
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1239720643
##
connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java:
##
@@ -192,6 +209,236 @@ public void testCancelAfterAwaitFlush() throws Exception {
flushFuture.get(1000, TimeUnit.MILLISECONDS);
}
+@Test
+public void
testFlushFailureWhenWriteToSecondaryStoreFailsForTombstoneOffsets() throws
InterruptedException, TimeoutException, ExecutionException {
Review Comment:
```suggestion
public void
testFlushFailureWhenWriteToSecondaryStoreFailsForTombstoneOffsets() throws
Exception {
```
nit: since this is just a test method, we can avoid the verbosity
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,59 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+Future secondaryWriteFuture = secondaryStore.set(values,
(secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteError.compareAndSet(null,
secondaryWriteError);
+} else {
+log.debug("Successfully flushed
tombstone(s)-containing to secondary backing store");
+}
+}
+});
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
+// tombstone records fail to be written to the secondary
store. Note that while commitTransaction
+// already waits for all records to be sent and ack'ed, in
this case we do need to add an explicit
+// blocking call. In case EOS is disabled, we wait for the
same duration as `offset.commit.timeout.ms`
+// and throw that exception which would allow the offset
commit to fail.
+if (exactlyOnce) {
+secondaryWriteFuture.get();
+} else {
+secondaryWriteFuture.get(offsetFlushTimeoutMs,
TimeUnit.MILLISECONDS);
+}
+} catch (InterruptedException e) {
+log.warn("{} Flush of tombstone(s)-containing to secondary
store interrupted, cancelling", this);
Review Comment:
`Flush of tombstone(s)-containing to secondary store`
Should this be `Flush of tombstone(s)-containing offsets batch to secondary
store` instead? 😅
Also I'm wondering if all of these log lines need to necessarily contain
this verbiage around tombstone containing offsets? Users probably don't really
need to know (or care) that offset flushes are handled differently internally
depending on whether or not they contain one or more `null` valued offsets.
IMO, just including the details of the failure should be sufficient - WDYT?
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,59 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+Future secondaryWriteFuture = secondaryStore.set(values,
(secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteE
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1255451696
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,10 +284,61 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+Future secondaryWriteFuture = secondaryStore.set(values,
(secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteError.compareAndSet(null,
secondaryWriteError);
+} else {
+log.debug("Successfully flushed tombstone offsets to
secondary backing store");
+}
+}
+});
Review Comment:
I was suggesting the use of a plain `FutureCallback` instantiated via the
default constructor. We're still creating an unnecessary callback to handle
`secondaryWriteError` which will anyway be surfaced synchronously when we call
`get` on our `FutureCallback`?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1255456909
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,60 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new
FutureCallback<>((secondaryWriteError, ignored) -> {
Review Comment:
Why do we need this underlying callback to handle `secondaryWriteError`?
Can't we just use a plain `FutureCallback` instantiated via the default
constructor ? The error will anyway be surfaced synchronously when we call
`get` on our `FutureCallback` (and handled in the `ExecutionException` catch
block)?
(Also mentioned in
https://github.com/apache/kafka/pull/13801#discussion_r1255451696, but that
comment was resolved and I don't seem to have permissions to un-resolve it).
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,60 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new
FutureCallback<>((secondaryWriteError, ignored) -> {
+try (LoggingContext context = loggingContext()) {
+if (secondaryWriteError != null) {
+log.warn("Failed to write offsets with tombstone
records to secondary backing store", secondaryWriteError);
+secondaryStoreTombstoneWriteError.compareAndSet(null,
secondaryWriteError);
+} else {
+log.debug("Successfully flushed
tombstone(s)-containing to secondary backing store");
Review Comment:
Same comment as https://github.com/apache/kafka/pull/13801/files#r1239708472
(also looks like that wasn't resolved either, even though the comment was
resolved).
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1258189389
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
Review Comment:
Can we also add a sentence or two explaining the reasoning here (wording can
be borrowed from https://issues.apache.org/jira/browse/KAFKA-15018)?
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
Review Comment:
This doesn't need to be an `AtomicReference` anymore
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new FutureCallback<>();
+secondaryStore.set(values, secondaryWriteFuture);
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
+// tombstone records fail to be written to the secondary
store. Note that while commitTransaction
+// already waits for all records to be sent and ack'ed, in
this case we do need to add an explicit
+// blocking call. In case EOS is disabled, we wait for the
same duration as `offset.commit.timeout.ms`
+// and throw that exception which would allow the offset
commit to fail.
+if (exactlyOnce) {
+secondaryWriteFuture.get();
+} else {
+secondaryWriteFuture.get(offsetFlushTimeoutMs,
TimeUnit.MILLISECONDS);
+}
+} catch (InterruptedException e) {
+log.warn("{} Flush of tombstone(s)-containing offsets to
secondary store interrupted, cancelling", this);
Review Comment:
Can we unify the language used in the log lines here and the subsequent
catch blocks? `tombstone(s)-containing offsets` / `offsets with tombstones`
and `secondary store` / `secondary storage`. Also, since most users won't be
aware of what "primary" and "secondary" stores are, I'd prefer using something
like worker / global offsets topic instead. As per
https://kafka.apache.org/coding-guide.html, logging is part of our "UI" 😄
##
connect/runtime/src/test/java/org/apache/kafka/connect/storage/OffsetStorageWriterTest.java:
##
@@ -192,6 +212,251 @@ public void testCancelAfterAwaitFlush() throws Exception {
flushFuture.get(1000, TimeUnit.MILLISECONDS);
}
+@Test
+public void
testFlushFailureWhenWriteToSecondaryStoreFailsForTombstoneOffsets() throws
Exception {
+
+KafkaOffsetBackingStore connectorStore =
setupOffsetBackingStoreWithProducer("topic1", false);
+KafkaOffsetBackingStore workerStore =
setupOffsetBackingStoreWithProducer("topic2", true);
+
+extractKeyValue(OFFSET_KEY, OFFSET_KEY_SERIALIZED, null, null);
+
+ConnectorOffsetBackingStore offsetBackingStore =
ConnectorOffsetBackingStore.withConnectorAndWorkerStores(
+() -> LoggingContext.forConnector("source-connector"),
+
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1259160336
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
Review Comment:
Aren't we still making explicit null checks here -
https://github.com/apache/kafka/pull/13801/files#diff-0014cfa4c32bfde215cb10bee987e997d5182815fcf1a1245539375519f83f3dR325-R326?
🤔
I'm fine with retaining it though, this isn't a blocking comment.
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
Review Comment:
Aren't we still making explicit null checks here -
https://github.com/apache/kafka/pull/13801/files#diff-0014cfa4c32bfde215cb10bee987e997d5182815fcf1a1245539375519f83f3dR325-R326?
🤔
I'm fine with retaining it though, this isn't a blocking comment.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1259161121
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new FutureCallback<>();
+secondaryStore.set(values, secondaryWriteFuture);
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
+// tombstone records fail to be written to the secondary
store. Note that while commitTransaction
+// already waits for all records to be sent and ack'ed, in
this case we do need to add an explicit
+// blocking call. In case EOS is disabled, we wait for the
same duration as `offset.commit.timeout.ms`
+// and throw that exception which would allow the offset
commit to fail.
+if (exactlyOnce) {
+secondaryWriteFuture.get();
+} else {
+secondaryWriteFuture.get(offsetFlushTimeoutMs,
TimeUnit.MILLISECONDS);
+}
+} catch (InterruptedException e) {
+log.warn("{} Flush of tombstone(s)-containing offsets to
secondary store interrupted, cancelling", this);
Review Comment:
Thanks, that makes sense 👍
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new FutureCallback<>();
+secondaryStore.set(values, secondaryWriteFuture);
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
+// tombstone records fail to be written to the secondary
store. Note that while commitTransaction
+// already waits for all records to be sent and ack'ed, in
this case we do need to add an explicit
+// blocking call. In case EOS is disabled, we wait for the
same duration as `offset.commit.timeout.ms`
+// and throw that exception which would allow the offset
commit to fail.
+if (exactlyOnce) {
+secondaryWriteFuture.get();
+} else {
+secondaryWriteFuture.get(offsetFlushTimeoutMs,
TimeUnit.MILLISECONDS);
+}
+} catch (InterruptedException e) {
+log.warn("{} Flush of tombstone(s)-containing offsets to
secondary store interrupted, cancelling", this);
Review Comment:
Thanks, that makes sense 👍
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1259168414
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new FutureCallback<>();
+secondaryStore.set(values, secondaryWriteFuture);
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
Review Comment:
> Well the rationale it the same that EOS waits indefinitely for the commits
to happen
Prior to this change, we just waited for the producer transaction commit to
complete since we didn't care about the writes to the secondary store (and the
transaction would contain all the data records as well as offset records
written to the connector's primary store). However, that would be bounded by
the producer's `transaction.timeout.ms` right?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] yashmayya commented on a diff in pull request #13801: KAFKA-15018: Failing offset flush for EOS when secondary offset store writes fails for tombstone records
yashmayya commented on code in PR #13801:
URL: https://github.com/apache/kafka/pull/13801#discussion_r1259168414
##
connect/runtime/src/main/java/org/apache/kafka/connect/storage/ConnectorOffsetBackingStore.java:
##
@@ -279,8 +284,51 @@ public Future set(Map
values, Callback callb
throw new IllegalStateException("At least one non-null offset
store must be provided");
}
+boolean containsTombstones = values.containsValue(null);
+
+// If there are tombstone offsets, then the failure to write to
secondary store will
+// not be ignored. Also, for tombstone records, we first write to
secondary store and
+// then to primary stores.
+if (secondaryStore != null && containsTombstones) {
+AtomicReference secondaryStoreTombstoneWriteError = new
AtomicReference<>();
+FutureCallback secondaryWriteFuture = new FutureCallback<>();
+secondaryStore.set(values, secondaryWriteFuture);
+try {
+// For EOS, there is no timeout for offset commit and it is
allowed to take as much time as needed for
+// commits. We still need to wait because we want to fail the
offset commit for cases when
Review Comment:
> Well the rationale it the same that EOS waits indefinitely for the commits
to happen
Prior to this change, we just waited for the producer transaction commit to
complete since we didn't care about the writes to the secondary offsets store
(and the transaction would contain all the data records as well as offset
records written to the connector's primary offsets store). However, that would
be bounded by the producer's `transaction.timeout.ms` right?
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
