https://pulsar.apache.org/ logo
Join Slack
Powered by
Another question on the Java Admin API. I have a t...
# general
g
Another question on the Java Admin API. I have a test where I wanted to check whether my code is acknowledging messages (integration test). I produce two messages to a topic, consume them without acking, and then consume them with the same named subscription again. I receive the same 2 messages in the second case. This is ok. But if I retrieve the number of unacked messages with:
Copy code
def getTopicStats(serviceUrl: String, topic: String) = {
  val admin = PulsarAdmin
    .builder()
    .serviceHttpUrl(serviceUrl)
    .build()
  try {
    admin.topics().getStats(topic)
  } finally {
    admin.close()
  }
}

val stats = getTopicStats(s"<http://localhost>:${pulsarServer.get.get.mappedPort(8080)}", topic.name)
stats.getSubscriptions.values.asScala.foreach { subscription =>
    subscription.getUnackedMessages shouldBe 0
}
Then it's 0, when I would expect some unacked messages to be there (reasoning, I expect the above test to fail if the acking is not done). If I use this instead:
Copy code
stats.getBacklogSize shouldBe 0
Then I get a test failure (when acking is done, the test passes).
l
unacked messages are ones that have been sent to consumers, but haven't been acked. Are you doing the assert after you have created the consumers?
g
Yes
Something like this (Scala code):
Copy code
def getResources[F[_]](pulsarClient: PulsarClient[F], topic: Topic) = {
  for {
    producer <- pulsarClient.makeProducer[String](topic, Schema.STRING)
    consumer <- pulsarClient.makeConsumer[String](
      topic,
      Schema.STRING,
      Subscription(subscriptionName)
    )
  } yield (producer, consumer)
}

  def simpleProduceConsumeTest(pulsarClient: PulsarClient[IO], topic: Topic, messagesToSend: List[String]) = {
    getResources(pulsarClient, topic)
      .use {
        case (producer, consumer) =>
          messagesToSend.traverse(producer.produce) *>
            PulsarClient.consume(consumer, processEvent).take(messagesToSend.size.toLong).compile.toList.map {
              messages =>
                messages shouldBe messagesToSend
            }
      }
      .unsafeRunSync()
  }

  def getTopicStats(serviceUrl: String, topic: String) = {
    val admin = PulsarAdmin
      .builder()
      .serviceHttpUrl(serviceUrl)
      .build()
    try {
      admin.topics().getStats(topic)
    } finally {
      admin.close()
    }
  }

  def assertBacklogIsEmpty() = {
    // Make an additional check with the Pulsar Admin API.
    val stats = getTopicStats(s"<http://localhost>:${pulsarServer.get.get.mappedPort(8080)}", topic.name)
    // The backlog size should be 0, as all messages were consumed.
    // Using this instead of checking the getUnackedMessages on each subscription, as the latter doesn't seem to work.
    stats.getBacklogSize shouldBe 0
  }

    "acknowledge messages after processing" in {
      val pulsarClient = PulsarClient[IO](s"<pulsar://localhost>:${pulsarServer.get.get.mappedPort(pulsarPort)}")

      simpleProduceConsumeTest(pulsarClient, topic, List("test message", "test message2"))

      simpleProduceConsumeTest(pulsarClient, topic, List("test message3", "test message4"))

      assertBacklogIsEmpty()
    }
pulsarClient is our own wrapper around the Pulsar Java API
The above code works (ie. catches missing acks). If I change the assertBacklogIsEmpty to check for the number of unacked messages (getUnackedMessages) for each subscription then I get 0 whether I really do acks, or not.
In other words, if I use 
getBacklogSize
, then my test is not prone to mutation (removing the ack functionality), but if I use 
getUnackedMessages
, then my test is prone to mutation.
2 Views
Open in Slack
PreviousNext
Powered by