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/12/10 17:25:58 UTC
[GitHub] [kafka] chia7712 opened a new pull request #9728: MINOR: make sure all dir jobs are completed
chia7712 opened a new pull request #9728:
URL: https://github.com/apache/kafka/pull/9728
revert the change introduced by #9680 and add a unit test to verify
### 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] chia7712 commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540483941
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +680,15 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
Review comment:
@kowshik Thanks for you explanation! It is indeed hard to give full test on the shutdown path.
> This test may not still catch the bug introduced in #9680.
you are right. I'm going to enrich the unit tests.
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540486925
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -435,6 +436,20 @@ class LogManager(logDirs: Seq[File],
cleaner.startup()
}
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @return true if all pass. Otherwise, false
+ */
+ private[log] def allPass(jobs: Seq[Future[_]]): Boolean = {
Review comment:
> I would even set the type parameter to the type we expect from the shutdown method.
There is no explicit type so I didn't define the type parameter.
----------------------------------------------------------------
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] ijuma commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540467496
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -435,6 +436,20 @@ class LogManager(logDirs: Seq[File],
cleaner.startup()
}
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @return true if all pass. Otherwise, false
+ */
+ private[log] def allPass(jobs: Seq[Future[_]]): Boolean = {
Review comment:
The log message talks explicitly about LogManager shutdown. Seems like you'd want it to say something about log manager shutdown. I would even set the type parameter to the type we expect from the shutdown method.
----------------------------------------------------------------
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] ijuma commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540461670
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -435,6 +436,20 @@ class LogManager(logDirs: Seq[File],
cleaner.startup()
}
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @return true if all pass. Otherwise, false
+ */
+ private[log] def allPass(jobs: Seq[Future[_]]): Boolean = {
Review comment:
The name makes this look generic, but it's not. The logging is very specific. I would change the name to be specific.
----------------------------------------------------------------
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] kowshik commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540478009
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +680,15 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
Review comment:
This test may not still catch the bug introduced in #9680.
We can continue more discussion under this review comment in #9680: https://github.com/apache/kafka/pull/9680/files#r540476157.
We should still get this PR merged though, since it fixes the problem.
cc @ijuma
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r552051062
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +681,31 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
+ def testWaitForAllToComplete(): Unit = {
+ var invokedCount = 0
+ val success: Future[Boolean] = Mockito.mock(classOf[Future[Boolean]])
+ Mockito.when(success.get()).thenAnswer(_ => invokedCount += 1)
Review comment:
```waitForAllToComplete``` does not use the return value so it does not cause casting error even though we don't return ```true```. However, it does look like a bug so I will add the return value.
----------------------------------------------------------------
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] chia7712 commented on pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#issuecomment-754635291
@ijuma Could you take 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540478009
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +680,15 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
Review comment:
This test may not still catch the bug introduced in #9680.
We can continue more discussion under this review comment in #9680: https://github.com/apache/kafka/pull/9680/files#r540476157.
We should still get this PR merged though, since it fixes the problem.
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -478,17 +493,8 @@ class LogManager(logDirs: Seq[File],
}
try {
- for ((dir, dirJobs) <- jobs) {
- val hasErrors = dirJobs.exists { future =>
- Try(future.get) match {
- case Success(_) => false
- case Failure(e) =>
- warn(s"There was an error in one of the threads during LogManager shutdown: ${e.getCause}")
- true
- }
- }
-
- if (!hasErrors) {
+ jobs.forKeyValue { (dir, dirJobs) =>
+ if (allPass(dirJobs)) {
Review comment:
This refactoring is good.
----------------------------------------------------------------
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] kowshik commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
kowshik commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540684755
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +680,15 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
Review comment:
Sounds good!
----------------------------------------------------------------
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] ijuma commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r552037417
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1167,6 +1160,21 @@ class LogManager(logDirs: Seq[File],
object LogManager {
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @param callback handle the exception caused by Future#get
Review comment:
We should mention that this will be called for each future that throws an exception.
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +681,31 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
+ def testWaitForAllToComplete(): Unit = {
+ var invokedCount = 0
+ val success: Future[Boolean] = Mockito.mock(classOf[Future[Boolean]])
+ Mockito.when(success.get()).thenAnswer(_ => invokedCount += 1)
Review comment:
Should we not return `true` in this answer?
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -1167,6 +1160,21 @@ class LogManager(logDirs: Seq[File],
object LogManager {
+ /**
+ * wait all jobs to complete
Review comment:
Nit: start with capital?
----------------------------------------------------------------
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] ijuma commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r543396355
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -435,6 +436,20 @@ class LogManager(logDirs: Seq[File],
cleaner.startup()
}
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @return true if all pass. Otherwise, false
+ */
+ private[log] def waitForAllToComplete(jobs: Seq[Future[_]]): Boolean = {
Review comment:
I would add a callback here for the failures and do the logging in the callback. That way you can keep the name generic.
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540463850
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -435,6 +436,20 @@ class LogManager(logDirs: Seq[File],
cleaner.startup()
}
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @return true if all pass. Otherwise, false
+ */
+ private[log] def allPass(jobs: Seq[Future[_]]): Boolean = {
Review comment:
How about "waitForAllToComplete"
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r543406537
##########
File path: core/src/main/scala/kafka/log/LogManager.scala
##########
@@ -435,6 +436,20 @@ class LogManager(logDirs: Seq[File],
cleaner.startup()
}
+ /**
+ * wait all jobs to complete
+ * @param jobs jobs
+ * @return true if all pass. Otherwise, false
+ */
+ private[log] def waitForAllToComplete(jobs: Seq[Future[_]]): Boolean = {
Review comment:
good suggestion. will address 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 merged pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 merged pull request #9728:
URL: https://github.com/apache/kafka/pull/9728
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9728: MINOR: make sure all dir jobs are completed
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9728:
URL: https://github.com/apache/kafka/pull/9728#discussion_r540484990
##########
File path: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
##########
@@ -680,4 +680,15 @@ class LogManagerTest {
time.sleep(logConfig.fileDeleteDelayMs + 1)
verifyMetrics(1)
}
+
+ @Test
Review comment:
For another, Your comment in ttps://github.com/apache/kafka/pull/9680/files#r540476157 is great to me. It would be better that we can make higher test coverage on that!
----------------------------------------------------------------
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