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
