Package net.bugreaper.modules.kafka

Class Kafka

java.lang.Object
net.bugreaper.modules.kafka.setup.KafkaConsumerHelper
net.bugreaper.modules.kafka.Kafka
All Implemented Interfaces:
KafkaAsserts, KafkaConfig, KafkaInt
public class Kafka extends KafkaConsumerHelper implements KafkaInt, KafkaAsserts, KafkaConfig
Kafka helper that provides a common API for operating with Kafka.

It is recommended to use a single instance: Kafka kafka = Kafka.getInstance();

Uses a static consumer membership to reduce consumer rebalancing delays after reconnects.

Important: Consumed messages remain available after the test execution. It is recommended to purge messages between tests or use unique test data.

Consumed messages are limited by default to KafkaConsumerHelper.maxConsumedMessages (only the latest messages are grabbed). The limit can be changed using setMaxConsumeMessages(int) or configuration: modules.kafka.max-consumed-messages.

Consumed messages are returned in reverse order (newest messages first) by default to improve performance for fast tests.

Assertions use the default await timeout KafkaConsumerHelper.awaitMs and wait for the first message. The timeout can be changed using setAwaitMs(int) or configuration: modules.kafka.await.

For testing, it is recommended to use one partition per topic, but multiple partitions are supported.

By default, the consumer uses a static group ID. A unique consumer can be enabled using setUniqueConsumer(boolean) or configuration: modules.kafka.generate-unique-consumer.

Since:
1.0.0

  • Constructor Details

    • Kafka

      public Kafka(String bootStrapServer)
      Creates a Kafka helper with the specified bootstrap server.
      Parameters:
      bootStrapServer - kafka server (example: "my-kafka:9092")

    • Kafka

      public Kafka()
      Constructs a MongoDb client using YAML configuration.

      Loads configuration values from a YAML file.

      Default file: bugreaper.yml

      Custom file: using -DbugreaperEnv=test loads bugreaper-test.yml

       modules:
   kafka:
     url: localhost:9096
     await: 300 # (optional)
     max-consumed-messages: 5 # (optional)
     max-consumer-timeout: 700 # (optional)
     generate-unique-consumer: true # (optional)
     reverse-messages: true # (optional)

      Missing required keys will result in configuration errors. Missing optional keys will fall back to predefined defaults.

      Throws:
      IllegalArgumentException - if the configuration contains invalid values

  • Method Details

    • getInstance

      public static Kafka getInstance()
      Returns the instance of Kafka with config builder Kafka().

      This implementation is thread-safe using method-level synchronization.

      Returns:
      the shared instance of Kafka
      Throws:
      IllegalArgumentException - if the configuration contains invalid values
      See Also:

    • setAwaitMs

      public Kafka setAwaitMs(int awaitMs)
      Description copied from interface: KafkaConfig
      Configures the global await timeout for assertions and operations that use await.
      Specified by:
      setAwaitMs in interface KafkaConfig
      Parameters:
      awaitMs - await timeout in milliseconds
      Returns:
      this instance for method chaining

    • setMaxConsumeMessages

      public Kafka setMaxConsumeMessages(int maxMessages)
      Description copied from interface: KafkaConfig
      Sets the maximum number of latest messages to consume.
      Specified by:
      setMaxConsumeMessages in interface KafkaConfig
      Parameters:
      maxMessages - maximum number of messages to consume from the latest offset
      Returns:
      this instance for method chaining

    • setConsumerTimeoutMs

      public Kafka setConsumerTimeoutMs(int consumerTimeoutMs)
      Description copied from interface: KafkaConfig
      Sets the consumer timeout.
      Specified by:
      setConsumerTimeoutMs in interface KafkaConfig
      Parameters:
      consumerTimeoutMs - maximum consumer execution time in milliseconds before interruption
      Returns:
      this instance for method chaining

    • setUniqueConsumer

      public Kafka setUniqueConsumer(boolean unique)
      Description copied from interface: KafkaConfig
      Enables or disables generating a unique consumer group ID.
      Specified by:
      setUniqueConsumer in interface KafkaConfig
      Parameters:
      unique - true to generate a unique consumer group ID, false to use the default group ID
      Returns:
      this instance for method chaining

    • setReverseMessages

      public Kafka setReverseMessages(boolean reverseMessages)
      Description copied from interface: KafkaConfig
      Sets the message ordering behavior.

      Controls whether consumed messages are returned in reverse order (newest messages first).

      Specified by:
      setReverseMessages in interface KafkaConfig
      Parameters:
      reverseMessages - true to return messages in reverse order, false to keep the original order
      Returns:
      this instance for method chaining

    • getConfigSummary

      public String getConfigSummary()
      Description copied from interface: KafkaConfig
      Returns and logs (at INFO level) a human-readable summary of all resolved configuration values.

      The summary includes values loaded from the YAML configuration file as well as any fields overridden programmatically after construction. Optional fields that were not present in the configuration and resolved via default values may also be included.

      Specified by:
      getConfigSummary in interface KafkaConfig
      Returns:
      String with summary

    • createTopic

      @Step("(Kafka) Create topic: {topic}") public void createTopic(String topic)
      Description copied from interface: KafkaInt
      Creates a topic with one partition.
      Specified by:
      createTopic in interface KafkaInt
      Parameters:
      topic - topic name

    • createTopic

      @Step("(Kafka) Create topic: \'{topic}\' with <{partitionsCount}> partitions") public void createTopic(String topic, int partitionsCount)
      Description copied from interface: KafkaInt
      Creates a topic with multiple partitions.
      Specified by:
      createTopic in interface KafkaInt
      Parameters:
      topic - topic name
      partitionsCount - number of partitions

    • deleteTopic

      @Step("(Kafka) Delete topic: \'{topic}\'") public void deleteTopic(String topic)
      Description copied from interface: KafkaInt
      Deletes a topic.
      Specified by:
      deleteTopic in interface KafkaInt
      Parameters:
      topic - topic name

    • purgeTopic

      @Step("(Kafka) Purge topic: \'{topic}\'") public void purgeTopic(String topic)
      Description copied from interface: KafkaInt
      Purges a topic by removing messages from all partitions.
      Specified by:
      purgeTopic in interface KafkaInt
      Parameters:
      topic - topic name

    • sendToTopic

      @Step("(Kafka) Send message to topic: \'{topic}\'") public void sendToTopic(String topic, String message)
      Description copied from interface: KafkaInt
      Sends a message to a topic.

      There may be a delay before the consumer can see the message, therefore await is used when consuming messages.

      Specified by:
      sendToTopic in interface KafkaInt
      Parameters:
      topic - topic name
      message - message content

    • sendToTopicWithKey

      @Step("(Kafka) Send message to topic: \'{topic}\' with key \'{key}\'") public void sendToTopicWithKey(String topic, String key, String message)
      Description copied from interface: KafkaInt
      Sends a message with a key to a topic.

      There may be a delay before the consumer can see the message

      Specified by:
      sendToTopicWithKey in interface KafkaInt
      Parameters:
      topic - topic name
      key - message key included in the Kafka record
      message - message content

    • grabMessagesFromTopic

      @Step("(Kafka) Grab messages from topic: \'{topic}\'") public net.bugreaper.core.assertable.AssertableStringList grabMessagesFromTopic(String topic)
      Description copied from interface: KafkaInt
      Grabs messages from a topic using a consumer without committing offsets.

      Uses await until at least one message exists in the topic.

      The maximum number of messages is configured globally (only the latest messages are grabbed).

      Messages are returned in reverse order (newest messages first) to optimize processing when working with large amounts of data.

      Uses static consumer membership to reduce consumer rebalancing delays after reconnecting.

      Example: grabMessagesFromTopic("test_queue").seeListAnyEquals("my message")

      Specified by:
      grabMessagesFromTopic in interface KafkaInt
      Parameters:
      topic - topic name
      Returns:
      AssertableStringList

    • grabMessagesFromTopic

      @Step("(Kafka) Grab messages from topic: \'{topic}\' with key: \'{key}\'") public net.bugreaper.core.assertable.AssertableStringList grabMessagesFromTopic(String topic, String key)
      Description copied from interface: KafkaInt
      Grabs messages from a topic using a consumer without committing offsets.

      Uses await until at least one message exists in the topic.

      The maximum number of messages is configured globally (only the latest messages are grabbed).

      Messages are returned in reverse order (newest messages first) to optimize processing when working with large amounts of data.

      Uses static consumer membership to reduce consumer rebalancing delays after reconnecting.

      Example: grabMessagesFromTopic("test_topic", "my-key").seeListAnyEquals("my message")

      Specified by:
      grabMessagesFromTopic in interface KafkaInt
      Parameters:
      topic - topic name
      key - message key to filter consumed records
      Returns:
      AssertableStringList

    • getAllTopicsNames

      @Step("(Kafka) Get all topics names") public net.bugreaper.core.assertable.AssertableStringList getAllTopicsNames()
      Description copied from interface: KafkaInt
      Returns a list of all topic names.

      Example: getAllTopicsNames().seeListAnyContains("test_topic")

      Specified by:
      getAllTopicsNames in interface KafkaInt
      Returns:
      AssertableStringList

    • getMessagesCountInTopic

      public int getMessagesCountInTopic(String topic)
      Description copied from interface: KafkaInt
      Returns the number of messages in a topic.
      Specified by:
      getMessagesCountInTopic in interface KafkaInt
      Parameters:
      topic - topic name
      Returns:
      number of messages in the topic

    • getPartitionsCount

      public int getPartitionsCount(String topic)
      Description copied from interface: KafkaInt
      Returns the number of partitions in a topic.
      Specified by:
      getPartitionsCount in interface KafkaInt
      Parameters:
      topic - topic name
      Returns:
      number of partitions in the topic

    • seeMessagesHaveEqualText

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' has a message EQUALS expected text") public void seeMessagesHaveEqualText(String topic, String expectedText)
      Description copied from interface: KafkaAsserts
      Asserts that at least one message in the topic is equal to the expected text.

      Uses await until at least one message exists.

      Specified by:
      seeMessagesHaveEqualText in interface KafkaAsserts
      Parameters:
      topic - topic name
      expectedText - expected message text

    • seeMessagesContainText

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' has a message CONTAINS expected text") public void seeMessagesContainText(String topic, String expectedPart)
      Description copied from interface: KafkaAsserts
      Asserts that at least one of messages in topic contains the specified text.

      Uses await until at least one message exists.

      Specified by:
      seeMessagesContainText in interface KafkaAsserts
      Parameters:
      topic - topic name
      expectedPart - expected text part

    • seeMessagesHaveEqualJson

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' has a message EQUALS expected JSON") public void seeMessagesHaveEqualJson(String topic, String expectedJson)
      Description copied from interface: KafkaAsserts
      Asserts that at least one of messages in topic is equal to the specified JSON with strict array ordering.

      Uses await until at least one message exists.

      Specified by:
      seeMessagesHaveEqualJson in interface KafkaAsserts
      Parameters:
      topic - topic name
      expectedJson - expected full JSON with strict array ordering

    • seeMessagesContainJson

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' has a message CONTAINS expected JSON") public void seeMessagesContainJson(String topic, String expectedJsonPart)
      Description copied from interface: KafkaAsserts
      Asserts that at least one of messages in topic contains the specified JSON without strict array ordering.

      Uses await until at least one message exists.

      Specified by:
      seeMessagesContainJson in interface KafkaAsserts
      Parameters:
      topic - topic name
      expectedJsonPart - expected JSON subset (array order is ignored, but the number of elements must match if provided)

    • seeMessagesCountInTopicExactly

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' contains EXACTLY <{expectedCount}> messages") public void seeMessagesCountInTopicExactly(String topic, int expectedCount)
      Description copied from interface: KafkaAsserts
      Asserts that the number of messages in the topic is exactly the expected count.

      Uses await.

      Specified by:
      seeMessagesCountInTopicExactly in interface KafkaAsserts
      Parameters:
      topic - topic name
      expectedCount - expected number of messages

    • seeTopicIsNotEmpty

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' is not empty") public void seeTopicIsNotEmpty(String topic)
      Description copied from interface: KafkaAsserts
      Asserts that the topic is not empty

      Uses await.

      Specified by:
      seeTopicIsNotEmpty in interface KafkaAsserts
      Parameters:
      topic - topic name

    • seeTopicIsEmpty

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' is empty") public void seeTopicIsEmpty(String topic)
      Description copied from interface: KafkaAsserts
      Asserts that the topic is empty.

      Uses await.

      Specified by:
      seeTopicIsEmpty in interface KafkaAsserts
      Parameters:
      topic - topic name

    • seeTopicExists

      @Step("(Kafka)[ASSERT] Topic: \'{topic}\' exists") public void seeTopicExists(String topic)
      Description copied from interface: KafkaAsserts
      Asserts that the topic exists.

      Uses await.

      Specified by:
      seeTopicExists in interface KafkaAsserts
      Parameters:
      topic - topic name