You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/08/04 10:06:57 UTC
[GitHub] [kafka] showuon opened a new pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl
showuon opened a new pull request #9121:
URL: https://github.com/apache/kafka/pull/9121
add 2 tests for IOExceptions cases:
1 for `OffsetCheckpoint` and 1 for `GlobalStateManagerImpl`
### 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-684518374
@mjsax , I didn't see the Jenkins build results. Should we run again? 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475362316
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
+
+ // checkpoint should fail due to the file is readonly
+ stateManager.checkpoint();
+ assertThat(appender.getMessages(), hasItem(containsString(
+ "Failed to write offset checkpoint file to " + checkpointFile.getPath() + " for global stores")));
Review comment:
No, we can't move the assert out of the try-block because the `appender` is declared within try block. We can move the assert out of try-block if we don't use the try resource auto-close pattern, but I don't think it would be better.
Also, we assert within try-block for the `LogCaptureAppender` in other places. I think they are all for the same reason as I mentioned 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax merged pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #9121:
URL: https://github.com/apache/kafka/pull/9121
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-701564709
Sorry for the delay. It's sometimes hard to review on time.
Thanks for the PR. Merged to `trunk`.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-696467921
@mjsax , sorry, I found you're online now. So could you take a look again for the long pending PR? It's 2nd review, so it should be easier. 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475393747
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
Review comment:
It's because we we will write data to the .tmp file first and then swap to the CHECKPOINT_FILE. And in the swap action, we use `Files.move` with `ATOMIC_MOVE` option, which will try to replace the target file if exists. I cannot create `IOException` case with this case. I added comments for this line to explain the reason. Thank you.
ref: https://docs.oracle.com/javase/8/docs/api/java/nio/file/Files.html#move-java.nio.file.Path-java.nio.file.Path-java.nio.file.CopyOption...-
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r477616241
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
Review comment:
Interesting... Thanks for clarification.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r477606207
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
+
+ // checkpoint should fail due to the file is readonly
+ stateManager.checkpoint();
+ assertThat(appender.getMessages(), hasItem(containsString(
+ "Failed to write offset checkpoint file to " + checkpointFile.getPath() + " for global stores")));
Review comment:
Ah. Good point. We can leave as-is.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon edited a comment on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon edited a comment on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-678959902
@mjsax , I've updated in this commit: https://github.com/apache/kafka/pull/9121/commits/12d3826a87a5b21033e2f81c6a486353e79d8591. 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-686345544
merge the latest trunk to have auto-build
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475393747
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
Review comment:
It's because we we will write data to the .tmp file first and then swap to the CHECKPOINT_FILE. And in the swap action, we use `Files.move` with `ATOMIC_MOVE` option, which will try to replace the target file if exists. I added comments for this line to explain the reason. Thank you.
ref: https://docs.oracle.com/javase/8/docs/api/java/nio/file/Files.html#move-java.nio.file.Path-java.nio.file.Path-java.nio.file.CopyOption...-
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-696467921
@mjsax , sorry, I found you're online now. So could you take a look again for the long pending PR? It's 2nd review, so it should be easier. 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-678959902
@mjsax , I've updated in this commit: https://github.com/apache/kafka/pull/9121/commits/b248ccc4f35ce6bb7ef865c65392710c1558ca20. 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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r474877105
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
Review comment:
Nit: no need to line-break
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
Review comment:
The test seems to be self-explaining and thus we don't really need this comment
##########
File path: streams/src/test/java/org/apache/kafka/streams/state/internals/OffsetCheckpointTest.java
##########
@@ -135,6 +136,20 @@ public void shouldThrowOnInvalidOffsetInWrite() throws IOException {
}
}
+ @Test
+ public void shouldThrowIOExceptionWhenWritingToNotExistedFile() {
+ final Map<TopicPartition, Long> offsetsToWrite = Collections.singletonMap(
+ new TopicPartition(topic, 0), 0L);
+
+ // create a file with not existed path, and feed into OffsetCheckpoint
Review comment:
that the pass is "not existing" is already clear from `File("not_existing_dir/not_existing_file")` and thus not necessary
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
+
+ // checkpoint should fail due to the file is readonly
+ stateManager.checkpoint();
+ assertThat(appender.getMessages(), hasItem(containsString(
+ "Failed to write offset checkpoint file to " + checkpointFile.getPath() + " for global stores")));
Review comment:
Subjective: Might be "better" to do the assertion after the try-block.
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
Review comment:
Why do we need to add `.tmp` ?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
+
+ // checkpoint should fail due to the file is readonly
Review comment:
as above; comment is not really needed
##########
File path: streams/src/test/java/org/apache/kafka/streams/state/internals/OffsetCheckpointTest.java
##########
@@ -135,6 +136,20 @@ public void shouldThrowOnInvalidOffsetInWrite() throws IOException {
}
}
+ @Test
+ public void shouldThrowIOExceptionWhenWritingToNotExistedFile() {
+ final Map<TopicPartition, Long> offsetsToWrite = Collections.singletonMap(
+ new TopicPartition(topic, 0), 0L);
Review comment:
nit: no need to line break
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-701733581
@mjsax , no problem at all. Totally understand 🙂
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-681148623
Retest this please.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475393747
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
Review comment:
It's because we we will write data to the .tmp file first and then swap to the CHECKPOINT_FILE. And in the swap action, we use `Files.move` with `ATOMIC_MOVE` option, which will try to replace the target file if exists. I cannot create `IOException` with this case. I added comments for this line to explain the reason. Thank you.
ref: https://docs.oracle.com/javase/8/docs/api/java/nio/file/Files.html#move-java.nio.file.Path-java.nio.file.Path-java.nio.file.CopyOption...-
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475393747
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
Review comment:
It's because we we will write data to the .tmp file first and then swap to the CHECKPOINT_FILE. And in the swap action, we use `Files.move` with `ATOMIC_MOVE` option, which will try to replace the target file if exists. I added comments for this line to explain the reason. Thank you.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
mjsax commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-668639088
Retest this please.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-668507540
@mjsax , could you review this PR to add tests? 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-689196185
@mjsax could you review this PR again? 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475358692
##########
File path: streams/src/test/java/org/apache/kafka/streams/state/internals/OffsetCheckpointTest.java
##########
@@ -135,6 +136,20 @@ public void shouldThrowOnInvalidOffsetInWrite() throws IOException {
}
}
+ @Test
+ public void shouldThrowIOExceptionWhenWritingToNotExistedFile() {
+ final Map<TopicPartition, Long> offsetsToWrite = Collections.singletonMap(
+ new TopicPartition(topic, 0), 0L);
+
+ // create a file with not existed path, and feed into OffsetCheckpoint
Review comment:
agree!
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-686864215
`tests/Build/JDK 11` and `tests/Build/JDK 15` test passed
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-668954076
**JDK11: failed with system failure**
* What went wrong:
03:15:27 Gradle build daemon disappeared unexpectedly (it may have been killed or may have crashed)
**JDK14: 2 failed tests are not related to my change**
1. kafka.network.SocketServerTest.testClientDisconnectionUpdatesRequestMetrics FAILED
2. org.apache.kafka.clients.consumer.StickyAssignorTest.testLargeAssignmentAndGroupWithUniformSubscription failed
**JDK8: pass**
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475362316
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
+
+ // checkpoint should fail due to the file is readonly
+ stateManager.checkpoint();
+ assertThat(appender.getMessages(), hasItem(containsString(
+ "Failed to write offset checkpoint file to " + checkpointFile.getPath() + " for global stores")));
Review comment:
I agree it'll be better to do the assertion after the try-block. But no, we can't move the assert out of the try-block because the `appender` is declared within try block. We can move the assert out of try-block if we don't use the try resource auto-close pattern, but I don't think it would be better.
Also, we assert within try-block for the `LogCaptureAppender` in other places. I think they are all for the same reason as I mentioned 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#discussion_r475362316
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java
##########
@@ -180,6 +183,27 @@ public void shouldReadCheckpointOffsets() throws IOException {
assertEquals(expected, offsets);
}
+ @Test
+ public void shouldLogWarningMessageWhenIOExceptionInCheckPoint() throws IOException {
+ final Map<TopicPartition, Long> offsets = Collections.singletonMap(t1, 25L);
+ stateManager.initialize();
+ stateManager.updateChangelogOffsets(offsets);
+
+ final File file = new File(stateDirectory.globalStateDir(), StateManagerUtil.CHECKPOINT_FILE_NAME + ".tmp");
+ file.createNewFile();
+ // set the checkpoint tmp file to read-only to simulate the IOException situation
+ file.setWritable(false);
+
+ try (final LogCaptureAppender appender =
+ LogCaptureAppender.createAndRegister(GlobalStateManagerImpl.class)) {
+
+ // checkpoint should fail due to the file is readonly
+ stateManager.checkpoint();
+ assertThat(appender.getMessages(), hasItem(containsString(
+ "Failed to write offset checkpoint file to " + checkpointFile.getPath() + " for global stores")));
Review comment:
I agree it'll be better to do the assertion after the try-block. But no, we can't move the assert out of the try-block because the `appender` is declared within try block. We can move the assert out of try-block if we don't use the try resource auto-close pattern, but I don't think it would be better.
Also, we assert within try-block for the `appender` tests in other places. I think they are all for the same reason as I mentioned 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-696467921
@mjsax , sorry, I found you're online now. So could you take a look again for the long pending PR? It's 2nd review, so it should be easier. 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-691990481
@mjsax could you review this PR again? 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:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #9121: KAFKA-10351: add tests for IOExceptions for GlobalStateManagerImpl/OffsetCheckpoint
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #9121:
URL: https://github.com/apache/kafka/pull/9121#issuecomment-675828368
@mjsax , could you review this small PR? 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:
users@infra.apache.org