You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jeqo (via GitHub)" <gi...@apache.org> on 2024/01/21 17:42:52 UTC
[PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
jeqo opened a new pull request, #15241:
URL: https://github.com/apache/kafka/pull/15241
Cached TransactionIndex may get closed if interrupted, causing following calls to always fail with ClosedChannelException, and forcing process to be restarted. In order to avoid this issue, a new method is exposed by TransactionIndex to validate state of channel; and index is reopened if closed.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2646846194
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1985844349
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.readFully(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
Review Comment:
That makes sense to me. @jeqo Will you please take a look at this? Perhaps we should revert this change and start again with the approach @junrao suggested.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon merged PR #15241:
URL: https://github.com/apache/kafka/pull/15241
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "nikramakrishnan (via GitHub)" <gi...@apache.org>.
nikramakrishnan commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2072530250
Bump! @satishd @kamalcph can we get this review going? 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1617724047
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -633,6 +657,10 @@ public String toString() {
", txnIndex=" + txnIndex.file().getName() +
'}';
}
+
+ public boolean hasTxnIndex() {
+ return txnIndex.file().length() == 0;
Review Comment:
is this check valid? should we have to invert it?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "jeqo (via GitHub)" <gi...@apache.org>.
jeqo commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1650469442
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -190,75 +175,145 @@ public void sanityCheck() {
AbortedTxn abortedTxn = txnWithPosition.txn;
if (abortedTxn.lastOffset() < startOffset)
throw new CorruptIndexException("Last offset of aborted transaction " + abortedTxn + " in index "
- + file.getAbsolutePath() + " is less than start offset " + startOffset);
+ + txnFile.path().toAbsolutePath() + " is less than start offset " + startOffset);
}
}
- private FileChannel openChannel() throws IOException {
- FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.CREATE,
- StandardOpenOption.READ, StandardOpenOption.WRITE);
- maybeChannel = Optional.of(channel);
- channel.position(channel.size());
- return channel;
- }
-
- private FileChannel channel() throws IOException {
- FileChannel channel = channelOrNull();
- if (channel == null)
- return openChannel();
- else
- return channel;
- }
-
- private FileChannel channelOrNull() {
- return maybeChannel.orElse(null);
- }
-
private Iterable<AbortedTxnWithPosition> iterable() {
return iterable(() -> ByteBuffer.allocate(AbortedTxn.TOTAL_SIZE));
}
private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate) {
- FileChannel channel = channelOrNull();
- if (channel == null)
+ if (!txnFile.exists())
return Collections.emptyList();
- PrimitiveRef.IntRef position = PrimitiveRef.ofInt(0);
+ try {
+ FileChannel channel = txnFile.channel();
+ PrimitiveRef.IntRef position = PrimitiveRef.ofInt(0);
+
+ return () -> new Iterator<AbortedTxnWithPosition>() {
+
+ @Override
+ public boolean hasNext() {
+ try {
+ return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ } catch (IOException e) {
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
+ }
+ }
+
+ @Override
+ public AbortedTxnWithPosition next() {
+ try {
+ ByteBuffer buffer = allocate.get();
+ Utils.readFully(channel, buffer, position.value);
+ buffer.flip();
+
+ AbortedTxn abortedTxn = new AbortedTxn(buffer);
+ if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
+ throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ + AbortedTxn.CURRENT_VERSION);
+ AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
+ position.value += AbortedTxn.TOTAL_SIZE;
+ return nextEntry;
+ } catch (IOException e) {
+ // We received an unexpected error reading from the index file. We propagate this as an
+ // UNKNOWN error to the consumer, which will cause it to retry the fetch.
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
+ }
+ }
- return () -> new Iterator<AbortedTxnWithPosition>() {
+ };
+
+ } catch (IOException e) {
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
+ }
+ }
- @Override
- public boolean hasNext() {
- try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
- } catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ public static class TransactionIndexFile {
Review Comment:
Seems that by making it private, it can only be accessed from TransactionIndex class but not tests or others on the same module.
Switching access to package-private and adding comment on visibility on last commit.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "jeqo (via GitHub)" <gi...@apache.org>.
jeqo commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1993215098
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.readFully(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
Review Comment:
@junrao Thanks for the suggestion and making the connection between the DelayedRemoteFetch and the RemoteLogReader getting interrupted. I was missing that.
I have a draft for the suggestion here: https://github.com/apache/kafka/pull/19197/commits/0f1332514234536bece94afc57f18f024174a9c6, let me know if it goes along the same lines.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "nikramakrishnan (via GitHub)" <gi...@apache.org>.
nikramakrishnan commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1509082067
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -522,6 +522,18 @@ public TimeIndex timeIndex() {
// Visible for testing
public TransactionIndex txnIndex() {
+ if (txnIndex.isClosed()) {
Review Comment:
Nit: Maybe it makes sense to add a debug log here to indicate that the log was closed and is being reopened? Might be useful to debug any performance issues.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2466582626
The patch LGTM. It needs second set of eyes as we are re-opening a closed index. cc @showuon @satishd PTAL.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1994293117
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.readFully(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
Review Comment:
Thanks for the draft, @jeqo. Left a few comments in https://github.com/apache/kafka/pull/19197.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2074186450
Thanks for the patch @jeqo!
I was able to reproduce the exception mentioned in the ticket by calling close in-middle of collecting the aborted transactions. The fix doesn't resolve the issue. Could you please write a unit test to ensure that the issue gets resolved?
https://github.com/apache/kafka/blob/trunk/storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java#L177
```
org.apache.kafka.common.KafkaException: Failed read position from the transaction index /var/folders/bq/w6tnkbq964q8sqpvj4fbmjr40000gq/T/kafka9502115127128961380.tmp
at org.apache.kafka.storage.internals.log.TransactionIndex$1.hasNext(TransactionIndex.java:245)
at org.apache.kafka.storage.internals.log.TransactionIndex.collectAbortedTxns(TransactionIndex.java:176)
at kafka.log.TransactionIndexTest.testCollectAbortedTxnsDuringClose(TransactionIndexTest.scala:210)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
...
Caused by: java.nio.channels.ClosedChannelException
at java.base/sun.nio.ch.FileChannelImpl.ensureOpen(FileChannelImpl.java:150)
at java.base/sun.nio.ch.FileChannelImpl.position(FileChannelImpl.java:325)
at org.apache.kafka.storage.internals.log.TransactionIndex$1.hasNext(TransactionIndex.java:243)
... 86 more
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2705463895
@jeqo , could you try to rebase to the latest trunk? I'd like to verify the change doesn't break the build. 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1985806237
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.readFully(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
Review Comment:
The solution is not ideal. When the channel is closed, it's possible that a LocalLog.read() is also reading from the channel and it will hit `ClosedByInterruptException`. Since this is an IOException, the `maybeHandleIOException()` logic will treat it as a failed disk and kill the broker (if there is only one disk).
I think it's better to change the logic in RemoteLogReader such that it never interrupts the thread. Note that we made other threads like ReplicaFetcherThread and CleanerThread non-interruptible for the same reason. So, this approach will be consistent with the existing ones.
Specially, we could change the following code to cancel without interruption.
` val cancelled = remoteFetchTask.cancel(true)`
We can then pass in a cancelled flag in RemoteLogReader. The flag will be set when the task is cancelled and `RemoteLogManager.read()` can check this flag to abort the operation early if needed. This way, we can avoid the risk of false disk failure and the overhead of reopening the file channel for txn index.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1985206261
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.readFully(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
Review Comment:
Having so many synchronized methods doesn't seem to follow the pattern we have used in the log layer. @junrao Thoughts?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "nikramakrishnan (via GitHub)" <gi...@apache.org>.
nikramakrishnan commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1509082067
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -522,6 +522,18 @@ public TimeIndex timeIndex() {
// Visible for testing
public TransactionIndex txnIndex() {
+ if (txnIndex.isClosed()) {
Review Comment:
Nit: Maybe it makes sense to add a debug log here to indicate that the index was closed and is being reopened? Might be useful to debug any performance issues.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2126963749
@jeqo , could you address @kamalcph 's comment above? 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2146529487
@jeqo
Any updates on this PR?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "jeqo (via GitHub)" <gi...@apache.org>.
jeqo commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2134015516
@kamalcph good catch! You're right. I have refactored the code a bit further to reopen the txn index properly. PTAL
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "jeqo (via GitHub)" <gi...@apache.org>.
jeqo commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2376067147
We have received feedback of this issue persisting on 3.8, so I guess we could have another look at this PR.
@showuon I have made the TransactionIndexFile channel access methods synchronized. Given that this is not an index expected to be concurrently accessed by many threads, this approach should be fine.
Also, got to know[1] that if the thread accessing the channel is interrupted, the channel will get closed implicitly. I think this is the main cause, as interruption are the way to handle delayed fetch operations. I have added a test to validate the channel gets reopen after interruption. PTAL.
[1]: https://stackoverflow.com/questions/1161297/why-are-we-getting-closedbyinterruptexception-from-filechannel-map-in-java-1-6/1161455#1161455
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1835574783
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.read(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
+ this.path = parentDir.resolve(path.getFileName());
+ }
+
+ synchronized void renameTo(Path other) throws IOException {
+ try {
+ if (Files.exists(path))
+ Utils.atomicMoveWithFallback(path, other, false);
+ } finally {
+ this.path = other;
+ }
+ }
+
+ synchronized void flush() throws IOException {
+ if (channel != null)
+ channel.force(true);
+ }
+
+ synchronized void closeChannel() throws IOException {
+ if (channel != null)
+ channel.close();
+ }
+
+ synchronized boolean isChannelOpen() {
+ return channel != null && channel.isOpen();
+ }
+
+ Path path() {
+ return path;
+ }
+
+ synchronized void truncate(long position) throws IOException {
+ if (channel != null)
+ channel.truncate(position);
+ }
+
+ boolean exists() {
+ return Files.exists(path);
+ }
+
+ boolean deleteIfExists() throws IOException {
+ closeChannel();
+ return Files.deleteIfExists(path());
+ }
+
+ void write(ByteBuffer buffer) throws IOException {
+ Utils.writeFully(channel(), buffer);
+ }
+
+ void read(ByteBuffer buffer, int position) throws IOException {
Review Comment:
can we rename this method to `readFully` as it doesn't read byte-by-byte?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "mjd95 (via GitHub)" <gi...@apache.org>.
mjd95 commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2376460752
We were the ones discussing with @jeqo - the "caching closed channels" issue was happening regularly for us on 3.8 in production, the thread doing a remote read was interrupted while iterating through the transaction index, we get a `ClosedByInterruptException` on some transaction index file channel, but the closed channel remains in the cache. The only way to mitigate was restarting the broker.
We were able to reproduce by setting a low `remote.fetch.max.wait.ms` and setting a small segment size in order to generate many transaction index files.
We tested that this PR fixes our repro and cherry-picked it onto our production release, we haven't seen the issue since then.
We haven't seen the "race during channel close" issue (which is now also handled in this PR) in production.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "jeqo (via GitHub)" <gi...@apache.org>.
jeqo commented on PR #15241:
URL: https://github.com/apache/kafka/pull/15241#issuecomment-2706210945
@showuon thanks! build looks good after rebase 👍🏽
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15931: Reopen TransactionIndex if channel is closed [kafka]
Posted by "jeqo (via GitHub)" <gi...@apache.org>.
jeqo commented on code in PR #15241:
URL: https://github.com/apache/kafka/pull/15241#discussion_r1988994745
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/TransactionIndex.java:
##########
@@ -238,35 +207,137 @@ private Iterable<AbortedTxnWithPosition> iterable(Supplier<ByteBuffer> allocate)
@Override
public boolean hasNext() {
try {
- return channel.position() - position.value >= AbortedTxn.TOTAL_SIZE;
+ return txnFile.currentPosition() - position.value >= AbortedTxn.TOTAL_SIZE;
} catch (IOException e) {
- throw new KafkaException("Failed read position from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed read position from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
@Override
public AbortedTxnWithPosition next() {
try {
ByteBuffer buffer = allocate.get();
- Utils.readFully(channel, buffer, position.value);
+ txnFile.readFully(buffer, position.value);
buffer.flip();
AbortedTxn abortedTxn = new AbortedTxn(buffer);
if (abortedTxn.version() > AbortedTxn.CURRENT_VERSION)
throw new KafkaException("Unexpected aborted transaction version " + abortedTxn.version()
- + " in transaction index " + file.getAbsolutePath() + ", current version is "
+ + " in transaction index " + txnFile.path().toAbsolutePath() + ", current version is "
+ AbortedTxn.CURRENT_VERSION);
AbortedTxnWithPosition nextEntry = new AbortedTxnWithPosition(abortedTxn, position.value);
position.value += AbortedTxn.TOTAL_SIZE;
return nextEntry;
} catch (IOException e) {
// We received an unexpected error reading from the index file. We propagate this as an
// UNKNOWN error to the consumer, which will cause it to retry the fetch.
- throw new KafkaException("Failed to read from the transaction index " + file.getAbsolutePath(), e);
+ throw new KafkaException("Failed to read from the transaction index " + txnFile.path().toAbsolutePath(), e);
}
}
};
}
+ // Visible for testing
+ static class TransactionIndexFile {
+ // note that the file is not created until we need it
+ private volatile Path path;
+ // channel is reopened as long as there are reads and writes
+ private FileChannel channel;
+
+ TransactionIndexFile(Path path) throws IOException {
+ this.path = path;
+
+ if (Files.exists(path))
+ openChannel();
+ }
+
+ private void openChannel() throws IOException {
+ channel = FileChannel.open(
+ path,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.READ,
+ StandardOpenOption.WRITE
+ );
+ channel.position(channel.size());
+ }
+
+ synchronized void updateParentDir(Path parentDir) {
Review Comment:
Sure, will have a look
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org