Migrate From Kafka Streams

This documentation covers the information you need to know to migrate an existing Kafka Streams application to Responsive. If you would like to see a sandboxed example of an end-to-end app, see the Quickstart.

Dependencies

Include the following dependencies into your project build.

Maven

<dependency>
  <groupId>dev.responsive</groupId>
  <artifactId>kafka-client</artifactId>
  <version>0.14.0</version>
</dependency>

Gradle

implementation 'dev.responsive:kafka-client:0.14.0'

Code Migration

There are three steps to migrate a Kafka Streams application to Responsive:

1. Migrate KafkaStreams
2. Migrate Stores
3. Update Configuration

Migrate KafkaStreams

Replace usages of new KafkaStreams with new ResponsiveKafkaStreams. Since ResponsiveKafkaStreams extends KafkaStreams, you should only need to change invocations of the constructor:

-   KafkaStreams streams = new KafkaStreams(...)
+   KafkaStreams streams = new ResponsiveKafkaStreams(...)
    // Everything else stays the same
    streams.start();

See the Kafka Streams API Reference for more details.

Migrate Stores

All methods in org.apache.kafka.streams.state.Stores have a corresponding method in ResponsiveStores to help make it ergonomic to replace your existing store usage/creation to Responsive. See some examples below for the different Kafka Streams APIs:

Using Stateful Operators (DSL)

success

We are currently implementing KIP-954 which will make migrating DSL applications to Responsive as simple as a single configuration change. Until then, follow this guide.

Avoid Unnecessary State

🦦 You may want to examine your original topology via Topology#describe in order to locate the exact operators which are connected to state stores; Streams may not always materialize state for all stateful operators. Adding Materialized where there was not previously a store is not incorrect, but it may cause unnecessary duplication of stored data that would otherwise be optimized away.

When using the DSL, state stores are either implicitly created as necessary or explicitly materialized via the Materialied class. You can create a Responsive store using ResponsiveStores#materialized:

  KTable<Long, Object> table = builder.table(
    topic,
-   Materialized.as(Stores.persistentKeyValueStore("table"),
+   ResponsiveStores.materialized(ResponsiveKeyValueParams.keyValue("table")
  );

Note that Materialized.as(ResponsiveStores.keyValueStore("table")) will compile but will not perform important validation steps that prevent errors. Make sure to always use the ResponsiveStores.materialized method.

Using Stateful Transformers (DSL + PAPI)

If you are using the DSL in conjunction with the PAPI, you must register stores manually with the StreamsBuilder. These stores should be modified to use ResponsiveStores:

  StreamsBuilder builder = new StreamsBuilder();
  builder.addStateStore(
-   Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(
+   ResponsiveStores.keyValueStoreBuilder(ResponsiveStores.keyValueStore(
      ...
    );
  );

  // this stays the same
  builder.process(() -> new Processor() {
    @Override public void init(final ProcessorContext context) {
      store = context.getStateStore(...);
    }
  }

Using Processor API (PAPI)

If you are using the PAPI, you must register stores manually with the Topology. These stores should be modified to use ResponsiveStores:

  Topology topology = new Topology();
  
  topology.addStateStore(
-   Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(...), ...),
+   ResponsiveStores.keyValueStoreBuilder(ResponsiveStores.keyValueStore(...), ...),
    "processor-1"
  );

  // this stays the same
  topology.addProcessor("processor-1", () -> new Processor(){
    @Override public void init(final ProcessorContext context) {
      store = context.getStateStore(...);
    }
  });

See the State Stores API Reference for more details, as well as some additional state store configurations that are exclusive to Responsive.

Update Configuration

When running your migrated Kafka Streams application with Responsive cloud as the backend, you will need to configure these additional properties. Simply add them to the Properties or config map that you pass in to the ResponsiveKafkaStreams constructor.

The values for all of these required configs will be provided to you:

responsive.storage.hostname=<RESPONSIVE STORAGE ENDPOINT>
responsive.storage.port=9042
responsive.storage.datacenter=AWS_US_WEST_2
responsive.tenant.id=<YOUR TENANT ID>
responsive.client.id=<API KEY>
responsive.client.secret=<API SECRET>

See Configuring your Application for more details.

Migrate Tests (Optional)

Responsive provides its own version of the TopologyTestDriver that can be used in exactly the same way, with all the same APIs, so that all you need to do is swap it out with ResponsiveTopologyTestDriver:

  Topology topology = appTopology();
  TopologyTestDriver testDriver = 
-   new TopologyTestDriver(topology, props, Instant.EPOCH);
+   new ResponsiveTopologyTestDriver(topology, props, Instant.EPOCH);

Like the TopologyTestDriver, the Responsive equivalent does not require any additional infrastructure such as a Kafka cluster or Responsive platform components such as the storage server or operator. The driver does run with real Responsive state stores; only the innermost layer that connects to the remote storage engine is swapped out for a simple in-memory map. This makes it a viable option for testing correctness and application logic.

note

You can check out ResponsiveTopologyTestDriverTest for some full examples of using the ResponsiveTopologyTestDriver if you are writing new tests from scratch.

Logging (Optional)

If supported by your logging implementation, you can configure Responsive logs to write directly to a Kafka Topic owned by Responsive. This will help us diagnose problems that your applications may encounter.

If you're using log4j, you can look at these example configurations to see how to set this up.

Supported Features

Responsive for Kafka Streams is intended to be a drop-in replacement for some of the classes bundled in with Kafka Streams. Some features are still under development. The table below gives a (non-comprehensive) overview of the current state of what is supported.

info

While some features may not be supported yet, Responsive is interoperable with deployed Kafka Streams applications. You may mix and match deployments of KafkaStreams and ResponsiveKafkaStreams, even within the same JVM.

Feature	Notes
Apache Kafka Version Compatibility	3.x
Key Value Store	range and all are not yet supported
Global Store	range and all are not yet supported
Window Stores	Support Coming Soon
Processing Guarantee	exactly_once_v2, at_least_once