Re: [PR] MINOR: Flaky ProducerIdManagerTest error injection fix [kafka]
soarez commented on PR #15605: URL: https://github.com/apache/kafka/pull/15605#issuecomment-2022452207 I think there is a JIRA already for this flaky test: [KAFKA-15915](https://issues.apache.org/jira/browse/KAFKA-15915) If this is correct, then you should update the PR title to refer to the JIRA. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINOR: Flaky ProducerIdManagerTest error injection fix [kafka]
viktorsomogyi commented on PR #15605: URL: https://github.com/apache/kafka/pull/15605#issuecomment-2022430965 @akatona84 I think this one exceeds the "minor" commit, I think you should create a jira ticket for this. Also, please add a description of your change that reveals the thinking behind your modifications. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] MINOR: Flaky ProducerIdManagerTest error injection fix [kafka]
akatona84 commented on code in PR #15605:
URL: https://github.com/apache/kafka/pull/15605#discussion_r1540748533
##
core/src/test/scala/unit/kafka/coordinator/transaction/ProducerIdManagerTest.scala:
##
@@ -38,19 +38,57 @@ import org.mockito.Mockito.{mock, when}
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.mutable
-import scala.util.{Failure, Success}
+import scala.util.{Failure, Success, Try}
class ProducerIdManagerTest {
var brokerToController: NodeToControllerChannelManager =
mock(classOf[NodeToControllerChannelManager])
val zkClient: KafkaZkClient = mock(classOf[KafkaZkClient])
+ case class ErrorCount(error: Errors, var repeat: Int)
+
+ object ErrorCount {
+val INDEFINITE: Int = -1
+
+def indefinitely(error: Errors): ErrorCount = {
+ ErrorCount(error, INDEFINITE)
+}
+ }
+
+ class ErrorQueue(initialErrorCounts: ErrorCount*) {
+private val queue: mutable.Queue[ErrorCount] = mutable.Queue.empty ++
initialErrorCounts
+
+def takeError(): Errors = queue.synchronized {
+ while (queue.head.repeat == 0) {
+queue.dequeue()
+ }
+ if (queue.head.repeat > 0) {
+queue.head.repeat -= 1
+ }
+ queue.head.error
+}
+
+def peekError(): Errors = queue.synchronized {
+ queue.head.error
+}
+
+def clearProcessedError(): Unit = {
+ TestUtils.waitUntilTrue(() =>
+queue.synchronized {
+ queue.head.repeat == 0
Review Comment:
I'm removing the INDEFINITE , empty queue will mean there's no error
(Errors.NONE is returned)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] MINOR: Flaky ProducerIdManagerTest error injection fix [kafka]
urbandan commented on code in PR #15605:
URL: https://github.com/apache/kafka/pull/15605#discussion_r1540691777
##
core/src/test/scala/unit/kafka/coordinator/transaction/ProducerIdManagerTest.scala:
##
@@ -38,19 +38,57 @@ import org.mockito.Mockito.{mock, when}
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.mutable
-import scala.util.{Failure, Success}
+import scala.util.{Failure, Success, Try}
class ProducerIdManagerTest {
var brokerToController: NodeToControllerChannelManager =
mock(classOf[NodeToControllerChannelManager])
val zkClient: KafkaZkClient = mock(classOf[KafkaZkClient])
+ case class ErrorCount(error: Errors, var repeat: Int)
+
+ object ErrorCount {
+val INDEFINITE: Int = -1
+
+def indefinitely(error: Errors): ErrorCount = {
+ ErrorCount(error, INDEFINITE)
+}
+ }
+
+ class ErrorQueue(initialErrorCounts: ErrorCount*) {
+private val queue: mutable.Queue[ErrorCount] = mutable.Queue.empty ++
initialErrorCounts
+
+def takeError(): Errors = queue.synchronized {
+ while (queue.head.repeat == 0) {
+queue.dequeue()
+ }
+ if (queue.head.repeat > 0) {
+queue.head.repeat -= 1
+ }
+ queue.head.error
+}
+
+def peekError(): Errors = queue.synchronized {
+ queue.head.error
+}
+
+def clearProcessedError(): Unit = {
+ TestUtils.waitUntilTrue(() =>
+queue.synchronized {
+ queue.head.repeat == 0
+}, "error wasn't processed")
+ queue.synchronized {
+queue.dequeue()
Review Comment:
shouldn't we clear the whole queue here?
##
core/src/test/scala/unit/kafka/coordinator/transaction/ProducerIdManagerTest.scala:
##
@@ -38,19 +38,57 @@ import org.mockito.Mockito.{mock, when}
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.mutable
-import scala.util.{Failure, Success}
+import scala.util.{Failure, Success, Try}
class ProducerIdManagerTest {
var brokerToController: NodeToControllerChannelManager =
mock(classOf[NodeToControllerChannelManager])
val zkClient: KafkaZkClient = mock(classOf[KafkaZkClient])
+ case class ErrorCount(error: Errors, var repeat: Int)
+
+ object ErrorCount {
+val INDEFINITE: Int = -1
+
+def indefinitely(error: Errors): ErrorCount = {
+ ErrorCount(error, INDEFINITE)
+}
+ }
+
+ class ErrorQueue(initialErrorCounts: ErrorCount*) {
+private val queue: mutable.Queue[ErrorCount] = mutable.Queue.empty ++
initialErrorCounts
+
+def takeError(): Errors = queue.synchronized {
+ while (queue.head.repeat == 0) {
+queue.dequeue()
+ }
+ if (queue.head.repeat > 0) {
+queue.head.repeat -= 1
+ }
+ queue.head.error
+}
+
+def peekError(): Errors = queue.synchronized {
+ queue.head.error
+}
+
+def clearProcessedError(): Unit = {
+ TestUtils.waitUntilTrue(() =>
+queue.synchronized {
+ queue.head.repeat == 0
Review Comment:
how is this going to behave if the background thread keeps calling
takeError, potentially removing all errors?
shouldn't the condition check for an empty queue?
also, what if the only element in the queue has INDEFINITE?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[PR] MINOR: Flaky ProducerIdManagerTest error injection fix [kafka]
akatona84 opened a new pull request, #15605: URL: https://github.com/apache/kafka/pull/15605 testUnrecoverableErrors was flaky as the wanted error either affected the next block request (prefecthing) or just missed that. ### 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
