인터랙티브 쿼리

인터랙티브 쿼리 (Interactive Queries)

카프카 스트림즈 애플리케이션이 관리하는 상태를 애플리케이션 외부에서 직접 조회하고 싶다면 인터랙티브 쿼리가 정답이에요. 상태 저장소를 마치 데이터베이스처럼 조회할 수 있게 해주는 기능이죠. 이 페이지에서는 로컬 상태 저장소를 쿼리하는 방법과, 전체 상태를 노출하고 싶을 때 필요한 RPC 레이어와 원격 조회까지 단계별로 설명해드릴게요.

출처: 문서

본문

인터랙티브 쿼리를 사용하면 애플리케이션 외부에서 애플리케이션의 상태를 활용할 수 있어요. Kafka Streams는 애플리케이션을 쿼리 가능하게(queryable) 만들어요.

애플리케이션의 전체 상태는 보통 많은 분산된 애플리케이션 인스턴스와, 이 애플리케이션 인스턴스들이 로컬로 관리하는 많은 상태 저장소에 걸쳐 분할돼요.

상태를 인터랙티브하게 쿼리하는 데는 로컬과 원격 구성 요소가 있어요.

• 로컬 상태 (Local state): 애플리케이션 인스턴스는 로컬로 관리하는 상태 부분을 쿼리하고 자신의 로컬 상태 저장소를 직접 쿼리할 수 있어요. Kafka Streams API를 호출할 필요가 없는 한, 해당 로컬 데이터를 애플리케이션 코드의 다른 부분에서 사용할 수 있어요. 상태 저장소를 쿼리하는 것은 항상 읽기 전용이라, 기본 상태 저장소가 대역 외(out-of-band)로 절대 변경되지 않도록 보장해요 (예: 새 항목을 추가할 수 없어요). 상태 저장소는 해당 프로세서 토폴로지와 그것이 처리하는 입력 데이터에 의해서만 변경되어야 해요. 자세한 내용은 [앱 인스턴스의 로컬 상태 저장소 쿼리하기](#을 참고해요.
• 원격 상태 (Remote state): 애플리케이션의 전체 상태를 쿼리하려면 상태의 다양한 조각들을 연결해야 해요:
  • 로컬 상태 저장소 쿼리
  • 네트워크에서 애플리케이션의 실행 중인 모든 인스턴스와 그 상태 저장소 발견
  • 네트워크를 통해 이 인스턴스들과 통신 (예: RPC 레이어)

이 조각들을 연결하면 같은 앱의 인스턴스 간 통신과, 인터랙티브 쿼리를 위한 다른 애플리케이션과의 통신이 가능해져요. 자세한 내용은 [전체 앱의 원격 상태 저장소 쿼리하기](#을 참고해요.

Kafka Streams는 애플리케이션의 상태를 인터랙티브하게 쿼리하는 데 필요한 모든 기능을 네이티브로 제공해요. 단, 애플리케이션의 전체 상태를 인터랙티브 쿼리로 노출하고 싶다면 예외예요. 애플리케이션 인스턴스가 네트워크를 통해 통신하도록 하려면 애플리케이션에 RPC(Remote Procedure Call) 레이어(예: REST API)를 추가해야 해요.

이 표는 다양한 절차에 대한 Kafka Streams 네이티브 통신 지원을 보여줘요.

절차	애플리케이션 인스턴스	전체 애플리케이션
앱 인스턴스의 로컬 상태 저장소 쿼리	지원됨	지원됨
앱 인스턴스를 다른 곳에서 발견 가능하게 만들기	지원됨	지원됨
실행 중인 모든 앱 인스턴스와 그 상태 저장소 발견	지원됨	지원됨
네트워크로 앱 인스턴스와 통신 (RPC)	지원됨	지원되지 않음 (직접 구성해야 함)

앱 인스턴스의 로컬 상태 저장소 쿼리

Kafka Streams 애플리케이션은 보통 여러 인스턴스에서 실행돼요. 특정 인스턴스에 로컬로 사용 가능한 상태는 애플리케이션 전체 상태의 부분 집합일 뿐이에요. 인스턴스의 로컬 저장소를 쿼리하면 해당 특정 인스턴스에 로컬로 사용 가능한 데이터만 반환해요.

KafkaStreams#store(...) 메서드는 이름과 유형으로 애플리케이션 인스턴스의 로컬 상태 저장소를 찾아요. 현재 이 시점에서 버전 상태 저장소(versioned state store)는 인터랙티브 쿼리가 지원되지 않는다는 점을 유의해요.

모든 애플리케이션 인스턴스는 자신의 로컬 상태 저장소를 직접 쿼리할 수 있어요. 상태 저장소의 이름은 저장소를 만들 때 정의돼요. Processor API를 사용해 명시적으로 만들거나 DSL의 상태 저장 연산으로 암묵적으로 만들 수 있어요.

상태 저장소의 유형은 QueryableStoreType으로 정의돼요. KafkaStreams#store(...)의 두 번째 인자로 QueryableStoreTypes의 내장 구현을 전달해요. 사용 가능한 내장 헬퍼:

• QueryableStoreTypes#keyValueStore() — 로컬 키-값 저장소 쿼리하기 참조.
• QueryableStoreTypes#timestampedKeyValueStore() — 로컬 키-값 저장소 쿼리하기 참조.
• QueryableStoreTypes#timestampedKeyValueStoreWithHeaders() — 헤더 인지 저장소와 인터랙티브 쿼리 참조.
• QueryableStoreTypes#windowStore() — 로컬 윈도우 저장소 쿼리하기 참조.
• QueryableStoreTypes#timestampedWindowStore() — 로컬 윈도우 저장소 쿼리하기 참조.
• QueryableStoreTypes#timestampedWindowStoreWithHeaders() — 헤더 인지 저장소와 인터랙티브 쿼리 참조.
• QueryableStoreTypes#sessionStore() — 로컬 윈도우 저장소 쿼리하기 참조.
• QueryableStoreTypes#sessionStoreWithHeaders() — 헤더 인지 저장소와 인터랙티브 쿼리 참조.

[로컬 커스텀 상태 저장소 쿼리하기](# 섹션에 설명된 대로 자신의 QueryableStoreType을 구현할 수도 있어요.

참고: Kafka Streams는 스트림 파티션당 하나의 상태 저장소를 구체화(materialize)해요. 즉 애플리케이션이 잠재적으로 많은 기본 상태 저장소를 관리할 거예요. API는 데이터가 어느 파티션에 있는지 알 필요 없이 모든 기본 저장소를 쿼리할 수 있게 해줘요.

참고: 헤더 인지 저장소의 경우, 인터랙티브 쿼리 결과에 레코드 헤더가 포함되어야 한다면 저장소 유형에 해당하는 위 목록의 *WithHeaders() 항목을 사용하세요.

로컬 키-값 저장소 쿼리

키-값 상태를 쿼리하려면 먼저 상태 저장소를 포함하는 토폴로지를 구성해야 해요. 이 예시는 그룹화된 스트림에 DSL count() 연산자를 사용하는데, CountsKeyValueStore라는 이름의 timestamped 키-값 저장소를 만들어요. 이 저장소는 word-count-input 토픽의 각 단어에 대한 최신 카운트를 보관해요.

Properties  props = ...;
StreamsBuilder builder = ...;
KStream<String, String> textLines = ...;

// Define the processing topology (here: WordCount)
KGroupedStream<String, String> groupedByWord = textLines
  .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
  .groupBy((key, word) -> word, Grouped.with(stringSerde, stringSerde));

// Create a key-value store named "CountsKeyValueStore" for the all-time word counts
groupedByWord.count(Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("CountsKeyValueStore"));

// Start an instance of the topology
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();

애플리케이션이 시작된 후 CountsKeyValueStore에 접근해 ReadOnlyKeyValueStore API를 통해 쿼리할 수 있어요:

// Get the key-value store CountsKeyValueStore
ReadOnlyKeyValueStore<String, Long> keyValueStore =
    streams.store("CountsKeyValueStore", QueryableStoreTypes.keyValueStore());

// Get value by key
System.out.println("count for hello:" + keyValueStore.get("hello"));

// Get the values for a range of keys available in this application instance
KeyValueIterator<String, Long> range = keyValueStore.range("all", "streams");
while (range.hasNext()) {
  KeyValue<String, Long> next = range.next();
  System.out.println("count for " + next.key + ": " + next.value);
}

// Get the values for all of the keys available in this application instance
KeyValueIterator<String, Long> range = keyValueStore.all();
while (range.hasNext()) {
  KeyValue<String, Long> next = range.next();
  System.out.println("count for " + next.key + ": " + next.value);
}

queryableStoreName을 받는 오버로드된 메서드를 사용해 무상태(stateless) 연산자의 결과를 구체화할 수도 있어요:

StreamsBuilder builder = ...;
KTable<String, Integer> regionCounts = ...;

// materialize the result of filtering corresponding to odd numbers
// the "queryableStoreName" can be subsequently queried.
KTable<String, Integer> oddCounts = numberLines.filter((region, count) -> (count % 2 != 0),
  Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as("queryableStoreName"));

// do not materialize the result of filtering corresponding to even numbers
// this means that these results will not be materialized and cannot be queried.
KTable<String, Integer> oddCounts = numberLines.filter((region, count) -> (count % 2 == 0));

로컬 윈도우 저장소 쿼리

윈도우 저장소는 키가 여러 윈도우에 존재할 수 있으므로 어떤 주어진 키에 대해 잠재적으로 많은 결과를 가질 수 있어요. 그러나 주어진 키에 대해 윈도우당 결과는 하나뿐이에요.

윈도우 저장소를 쿼리하려면 먼저 윈도우 집계로 토폴로지를 구성해야 해요 (예: windowedBy 다음에 count() 사용). 이 예시는 단어별 카운트를 위한 1분 윈도우로 CountsWindowStore라는 이름의 timestamped 윈도우 저장소를 만들기 위해 count()를 사용해요.

StreamsBuilder builder = ...;
KStream<String, String> textLines = ...;

// Define the processing topology (here: WordCount)
KGroupedStream<String, String> groupedByWord = textLines
  .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
  .groupBy((key, word) -> word, Grouped.with(stringSerde, stringSerde));

// Create a window state store named "CountsWindowStore" that contains the word counts for every minute
groupedByWord.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(60)))
  .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("CountsWindowStore"));

애플리케이션이 시작된 후 CountsWindowStore에 접근해 ReadOnlyWindowStore API를 통해 쿼리할 수 있어요:

// Get the window store named "CountsWindowStore"
ReadOnlyWindowStore<String, Long> windowStore =
    streams.store("CountsWindowStore", QueryableStoreTypes.windowStore());

// Fetch values for the key "world" for all of the windows available in this application instance.
// To get *all* available windows we fetch windows from the beginning of time until now.
Instant timeFrom = Instant.ofEpochMilli(0); // beginning of time = oldest available
Instant timeTo = Instant.now(); // now (in processing-time)
WindowStoreIterator<Long> iterator = windowStore.fetch("world", timeFrom, timeTo);
while (iterator.hasNext()) {
  KeyValue<Long, Long> next = iterator.next();
  long windowTimestamp = next.key;
  System.out.println("Count of 'world' @ time " + windowTimestamp + " is " + next.value);
}

로컬 커스텀 상태 저장소 쿼리

참고: 커스텀 상태 저장소는 Processor API만 지원해요.

커스텀 상태 저장소를 쿼리하기 전에 다음 인터페이스를 구현해야 해요:

• 사용자 커스텀 상태 저장소는 StateStore를 구현해야 해요.
• 저장소에서 사용 가능한 연산을 나타내는 인터페이스가 있어야 해요.
• 저장소 인스턴스를 만들기 위한 StoreBuilder의 구현을 제공해야 해요.
• 읽기 전용 연산으로 접근을 제한하는 인터페이스를 제공하는 것이 권장돼요. 이것은 이 API 사용자가 실행 중인 Kafka Streams 애플리케이션의 상태를 대역 외로 변경하는 것을 방지해요.

커스텀 저장소의 클래스/인터페이스 계층은 다음과 비슷할 수 있어요:

public class MyCustomStore<K,V> implements StateStore, MyWriteableCustomStore<K,V> {
  // implementation of the actual store
}

// Read-write interface for MyCustomStore
public interface MyWriteableCustomStore<K,V> extends MyReadableCustomStore<K,V> {
  void write(K Key, V value);
}

// Read-only interface for MyCustomStore
public interface MyReadableCustomStore<K,V> {
  V read(K key);
}

public class MyCustomStoreBuilder implements StoreBuilder {
  // implementation of the supplier for MyCustomStore
}

이 저장소를 쿼리 가능하게 만들려면:

• QueryableStoreType의 구현을 제공.
• 저장소의 모든 기본 인스턴스에 접근하고 쿼리에 사용되는 래퍼 클래스를 제공.

QueryableStoreType을 구현하는 방법:

public class MyCustomStoreType<K,V> implements QueryableStoreType<MyReadableCustomStore<K,V>> {

  // Only accept StateStores that are of type MyCustomStore
  public boolean accepts(final StateStore stateStore) {
    return stateStore instanceOf MyCustomStore;
  }

  public MyReadableCustomStore<K,V> create(final StateStoreProvider storeProvider, final String storeName) {
      return new MyCustomStoreTypeWrapper(storeProvider, storeName, this);
  }

}

Kafka Streams 애플리케이션의 각 인스턴스는 여러 스트림 태스크를 실행하고 특정 상태 저장소의 여러 로컬 인스턴스를 관리할 수 있으므로 래퍼 클래스가 필요해요. 래퍼 클래스는 이 복잡성을 숨기고, 저장소의 모든 기본 로컬 인스턴스에 대해 알 필요 없이 이름으로 "논리적" 상태 저장소를 쿼리할 수 있게 해줘요.

래퍼 클래스를 구현할 때는 저장소의 기본 인스턴스에 접근하기 위해 StateStoreProvider 인터페이스를 사용해야 해요. StateStoreProvider#stores(String storeName, QueryableStoreType<T> queryableStoreType)는 주어진 storeName과 queryableStoreType이 정의하는 유형의 상태 저장소 List를 반환해요.

래퍼 구현 예시:

// We strongly recommended implementing a read-only interface
// to restrict usage of the store to safe read operations!
public class MyCustomStoreTypeWrapper<K,V> implements MyReadableCustomStore<K,V> {

  private final QueryableStoreType<MyReadableCustomStore<K, V>> customStoreType;
  private final String storeName;
  private final StateStoreProvider provider;

  public CustomStoreTypeWrapper(final StateStoreProvider provider,
                              final String storeName,
                              final QueryableStoreType<MyReadableCustomStore<K, V>> customStoreType) {

    // ... assign fields ...
  }

  // Implement a safe read method
  @Override
  public V read(final K key) {
    // Get all the stores with storeName and of customStoreType
    final List<MyReadableCustomStore<K, V>> stores = provider.getStores(storeName, customStoreType);
    // Try and find the value for the given key
    final Optional<V> value = stores.stream().filter(store -> store.read(key) != null).findFirst();
    // Return the value if it exists
    return value.orElse(null);
  }

}

이제 커스텀 저장소를 찾아 쿼리할 수 있어요:

Topology topology = ...;
ProcessorSupplier processorSuppler = ...;

// Create CustomStoreSupplier for store name the-custom-store
MyCustomStoreBuilder customStoreBuilder = new MyCustomStoreBuilder("the-custom-store") //...;
// Add the source topic
topology.addSource("input", "inputTopic");
// Add a custom processor that reads from the source topic
topology.addProcessor("the-processor", processorSupplier, "input");
// Connect your custom state store to the custom processor above
topology.addStateStore(customStoreBuilder, "the-processor");

KafkaStreams streams = new KafkaStreams(topology, config);
streams.start();

// Get access to the custom store
MyReadableCustomStore<String,String> store = streams.store("the-custom-store", new MyCustomStoreType<String,String>());
// Query the store
String value = store.read("key");

전체 앱의 원격 상태 저장소 쿼리

전체 앱의 원격 상태를 쿼리하려면 애플리케이션의 전체 상태를 다른 애플리케이션(다른 머신에서 실행되는 애플리케이션 포함)에 노출해야 해요.

예를 들어, 멀티플레이어 비디오 게임에서 사용자 이벤트를 처리하는 Kafka Streams 애플리케이션이 있고, 각 사용자의 최신 상태를 직접 가져와 모바일 앱에 표시하고 싶다고 해볼게요. 애플리케이션의 전체 상태를 쿼리 가능하게 만드는 데 필요한 단계:

• 애플리케이션에 RPC 레이어를 추가해 애플리케이션의 인스턴스가 네트워크를 통해 상호작용할 수 있게 해요 (예: REST API, Thrift, 커스텀 프로토콜 등). 인스턴스는 인터랙티브 쿼리에 응답해야 해요. 참고 예시를 따라 시작할 수 있어요.
• Kafka Streams의 application.server 구성 설정을 통해 애플리케이션 인스턴스의 RPC 엔드포인트를 노출해요. RPC 엔드포인트는 네트워크 내에서 고유해야 하므로 각 인스턴스는 이 구성 설정에 대해 자신만의 값을 가져요. 이것은 애플리케이션 인스턴스를 다른 인스턴스가 발견할 수 있게 해줘요.
• RPC 레이어에서 원격 애플리케이션 인스턴스와 그 상태 저장소를 발견하고, 로컬로 사용 가능한 상태 저장소를 쿼리해 애플리케이션의 전체 상태를 쿼리 가능하게 만들어요. 특정 인스턴스가 쿼리에 응답할 로컬 데이터가 부족하면 원격 애플리케이션 인스턴스가 쿼리를 다른 앱 인스턴스로 전달할 수 있어요. 로컬로 사용 가능한 상태 저장소는 쿼리에 직접 응답할 수 있어요.

애플리케이션에 RPC 레이어 추가

RPC 레이어를 추가하는 방법은 여러 가지가 있어요. 유일한 요구사항은 RPC 레이어가 Kafka Streams 애플리케이션 내에 임베드되고, 다른 애플리케이션 인스턴스와 애플리케이션이 연결할 수 있는 엔드포인트를 노출한다는 것이에요.

애플리케이션의 RPC 엔드포인트 노출

분산 Kafka Streams 애플리케이션에서 원격 상태 저장소 발견을 활성화하려면 구성 속성에 구성 속성을 설정해야 해요. application.server 속성은 Kafka Streams 애플리케이션의 해당 인스턴스의 RPC 엔드포인트를 가리키는 고유한 host:port 쌍을 정의해요. 이 구성 속성의 값은 애플리케이션 인스턴스마다 달라져요. 이 속성이 설정되면 Kafka Streams는 StreamsMetadata 인스턴스를 통해 애플리케이션의 모든 인스턴스, 그 상태 저장소, 할당된 스트림 파티션에 대한 RPC 엔드포인트 정보를 추적해요.

팁: 노출된 RPC 엔드포인트를 인터랙티브 쿼리를 넘어서는 추가 인터-애플리케이션 통신(예: 추가 기능을 함께 실어 보내기)에 활용하는 것을 고려해보세요.

이 예시는 상태 저장소의 발견을 지원하는 Kafka Streams 애플리케이션을 구성하고 실행하는 방법을 보여줘요.

Properties props = new Properties();
// Set the unique RPC endpoint of this application instance through which it
// can be interactively queried.  In a real application, the value would most
// probably not be hardcoded but derived dynamically.
String rpcEndpoint = "host1:4460";
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, rpcEndpoint);
// ... further settings may follow here ...

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> textLines = builder.stream(stringSerde, stringSerde, "word-count-input");

final KGroupedStream<String, String> groupedByWord = textLines
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word, Grouped.with(stringSerde, stringSerde));

// This call to `count()` creates a state store named "word-count".
// The state store is discoverable and can be queried interactively.
groupedByWord.count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("word-count"));

// Start an instance of the topology
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();

// Then, create and start the actual RPC service for remote access to this
// application instance's local state stores.
//
// This service should be started on the same host and port as defined above by
// the property `StreamsConfig.APPLICATION_SERVER_CONFIG`.  The example below is
// fictitious, but we provide end-to-end demo applications (such as KafkaMusicExample)
// that showcase how to implement such a service to get you started.
MyRPCService rpcService = ...;
rpcService.listenAt(rpcEndpoint);

애플리케이션 인스턴스와 그 로컬 상태 저장소 발견 및 접근

다음 메서드는 애플리케이션 인스턴스에 대한 메타 정보(예: RPC 엔드포인트, 로컬로 사용 가능한 상태 저장소)를 제공하는 StreamsMetadata 객체를 반환해요.

• KafkaStreams#allMetadata(): 이 애플리케이션의 모든 인스턴스 찾기
• KafkaStreams#allMetadataForStore(String storeName): 상태 저장소 "storeName"의 로컬 인스턴스를 관리하는 애플리케이션 인스턴스 찾기
• KafkaStreams#metadataForKey(String storeName, K key, Serializer<K> keySerializer): 기본 스트림 파티셔닝 전략을 사용해, 주어진 상태 저장소에서 주어진 키의 데이터를 보유한 애플리케이션 인스턴스 하나 찾기
• KafkaStreams#metadataForKey(String storeName, K key, StreamPartitioner<K, ?> partitioner): partitioner를 사용해, 주어진 상태 저장소에서 주어진 키의 데이터를 보유한 애플리케이션 인스턴스 하나 찾기

주의: 애플리케이션 인스턴스에 대해 application.server가 구성되지 않으면 위 메서드들은 그것에 대해 어떤 StreamsMetadata도 찾지 못해요.

예를 들어, 이제 이전 섹션의 코드 예시에서 정의한 "word-count"라는 이름의 상태 저장소에 대한 StreamsMetadata를 찾을 수 있어요:

KafkaStreams streams = ...;
// Find all the locations of local instances of the state store named "word-count"
Collection<StreamsMetadata> wordCountHosts = streams.allMetadataForStore("word-count");

// For illustrative purposes, we assume using an HTTP client to talk to remote app instances.
HttpClient http = ...;

// Get the word count for word (aka key) 'alice': Approach 1
//
// We first find the one app instance that manages the count for 'alice' in its local state stores.
StreamsMetadata metadata = streams.metadataForKey("word-count", "alice", Serdes.String().serializer());
// Then, we query only that single app instance for the latest count of 'alice'.
// Note: The RPC URL shown below is fictitious and only serves to illustrate the idea.  Ultimately,
// the URL (or, in general, the method of communication) will depend on the RPC layer you opted to
// implement.  Again, we provide end-to-end demo applications (such as KafkaMusicExample) that showcase
// how to implement such an RPC layer.
Long result = http.getLong("http://" + metadata.host() + ":" + metadata.port() + "/word-count/alice");

// Get the word count for word (aka key) 'alice': Approach 2
//
// Alternatively, we could also choose (say) a brute-force approach where we query every app instance
// until we find the one that happens to know about 'alice'.
Optional<Long> result = streams.allMetadataForStore("word-count")
    .stream()
    .map(streamsMetadata -> {
        // Construct the (fictituous) full endpoint URL to query the current remote application instance
        String url = "http://" + streamsMetadata.host() + ":" + streamsMetadata.port() + "/word-count/alice";
        // Read and return the count for 'alice', if any.
        return http.getLong(url);
    })
    .filter(s -> s != null)
    .findFirst();

이 시점에서 애플리케이션의 전체 상태가 인터랙티브하게 쿼리 가능해요:

• 애플리케이션의 실행 중인 인스턴스와 그것들이 로컬로 관리하는 상태 저장소를 발견할 수 있어요.
• 애플리케이션에 추가된 RPC 레이어를 통해 네트워크로 이 애플리케이션 인스턴스와 통신하고 로컬로 사용 가능한 상태를 쿼리할 수 있어요.
• 애플리케이션 인스턴스는 자신의 로컬 상태 저장소를 직접 쿼리하고 RPC 레이어로 응답할 수 있으므로 그러한 쿼리를 제공할 수 있어요.
• 종합적으로, 이것은 전체 애플리케이션의 상태를 쿼리할 수 있게 해줘요.

인터랙티브 쿼리가 있는 엔드투엔드 애플리케이션을 보려면 데모 애플리케이션을 검토해보세요.

더 알아보기

• Streams 애플리케이션 구성하기 — application.server 등 관련 구성을 봐요.
• Streams 개발자 가이드 — 전체 개발 문서를 봐요.
• Processor API — 커스텀 상태 저장소를 만드는 맥락을 봐요.