Package com.bakdata.kafka

Class TestApplicationRunner

java.lang.Object
com.bakdata.kafka.TestApplicationRunner
public final class TestApplicationRunner extends Object
Class that provides helpers for using running KafkaApplication in tests

  • Method Details

    • create

      public static TestApplicationRunner create(String bootstrapServers)
      Create a test application runner with the given bootstrap servers.
      Parameters:
      bootstrapServers - bootstrap servers to connect to
      Returns:
      test application runner

    • withStateDir

      public TestApplicationRunner withStateDir(Path stateDir)
      Configure StreamsConfig.STATE_DIR_CONFIG for Kafka Streams. Useful for testing
      Parameters:
      stateDir - directory to use for storing Kafka Streams state
      Returns:
      a copy of this runtime configuration with configured Kafka Streams state directory

    • withNoStateStoreCaching

      public TestApplicationRunner withNoStateStoreCaching()
      Disable for Kafka Streams. Useful for testing
      Returns:
      a copy of this runtime configuration with Kafka Streams state store caching disabled

    • withSessionTimeout

      public TestApplicationRunner withSessionTimeout(Duration sessionTimeout)
      Configure ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG for Kafka consumers. Useful for testing
      Parameters:
      sessionTimeout - session timeout
      Returns:
      a copy of this runtime configuration with configured consumer session timeout

    • withKafkaConfig

      public TestApplicationRunner withKafkaConfig(Map<String,String> newKafkaConfig)
      Configure arbitrary Kafka properties
      Parameters:
      newKafkaConfig - properties to configure
      Returns:
      a copy of this runtime configuration with provided properties

    • withSchemaRegistry

      public TestApplicationRunner withSchemaRegistry(@NonNull @NonNull TestSchemaRegistry schemaRegistry)
      Configure a schema registry for (de-)serialization.
      Parameters:
      schemaRegistry - schema registry to use
      Returns:
      a copy of this runtime configuration with configured schema registry

    • withSchemaRegistry

      public TestApplicationRunner withSchemaRegistry()
      Configure a schema registry for (de-)serialization.
      Returns:
      a copy of this runtime configuration with configured schema registry

    • run

      public CompletableFuture<Integer> run(KafkaApplication<?,?,?,?,?,?,?,?> app, String... args)
      Run the application asynchronously with the given arguments. --bootstrap-servers, --schema -registry-url, and --kafka-config are automatically configured.
      Parameters:
      app - application to run
      args - CLI arguments to pass to the application
      Returns:
      CompletableFuture providing the application exit code

    • clean

      public int clean(KafkaApplication<?,?,?,?,?,?,?,?> app, String... args)
      Clean the application with the given arguments. --bootstrap-servers, --schema-registry-url, and --kafka-config are automatically configured.
      Parameters:
      app - application to clean
      args - CLI arguments to pass to the application
      Returns:
      application exit code

    • reset

      public int reset(KafkaStreamsApplication<? extends StreamsApp> app, String... args)
      Reset the application with the given arguments. --bootstrap-servers, --schema-registry-url, and --kafka-config are automatically configured.
      Parameters:
      app - application to reset
      args - CLI arguments to pass to the application
      Returns:
      application exit code

    • run

      public CompletableFuture<Void> run(KafkaApplication<?,?,?,?,?,?,?,?> app)
      Run the application asynchronously. Bootstrap servers, Schema Registry and Kafka config are automatically configured.
      Parameters:
      app - application to run
      Returns:
      CompletableFuture to await execution

    • clean

      public void clean(KafkaApplication<?,?,?,?,?,?,?,?> app)
      Clean the application. Bootstrap servers, Schema Registry and Kafka config are automatically configured.
      Parameters:
      app - application to clean

    • reset

      public void reset(KafkaStreamsApplication<? extends StreamsApp> app)
      Reset the application. Bootstrap servers, Schema Registry and Kafka config are automatically configured.
      Parameters:
      app - application to reset

    • verify

      public ConsumerGroupVerifier verify(KafkaStreamsApplication<? extends StreamsApp> app)
      Create a new ConsumerGroupVerifier for the provided application.
      Parameters:
      app - application to verify
      Returns:
      ConsumerGroupVerifier

    • newTestClient

      public KafkaTestClient newTestClient()
      Create a new KafkaTestClient for the configured test environment.
      Returns:
      KafkaTestClient

    • configure

      public void configure(KafkaApplication<?,?,?,?,?,?,?,?> app)
      Configure the application according to the test environment. Bootstrap servers, Schema Registry and Kafka config are configured.
      Parameters:
      app - application to configure