- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to - * the same key. - * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of - * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for - * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and - * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}. - *
- * To query the local {@link KeyValueStore} it must be obtained via - * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}: - *
{@code
- * KafkaStreams streams = ... // counting words
- * ReadOnlyKeyValueStore localStore = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
- * String key = "some-word";
- * Long countForWord = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
- * }
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param queryableStoreName the name of the underlying {@link KTable} state store; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} then this will be equivalent to {@link KGroupedStream#count()}.
- * @return a {@link KTable} that contains "update" records with unmodified keys and {@link Long} values that
- * represent the latest (rolling) count (i.e., number of records) for each key
- * @deprecated use {@link #count(Materialized) count(Materialized.as(queryableStoreName))}
- */
- @Deprecated
- KTable
* Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
@@ -130,10 +86,9 @@ public interface KGroupedStream
* To query the local {@link KeyValueStore} it must be obtained via
* {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStore#name()} to get the store name:
*
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
*
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param windows the specification of the aggregation {@link Windows}
- * @param queryableStoreName the name of the underlying {@link KTable} state store; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} then this will be equivalent to {@link KGroupedStream#count(Windows)}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys and {@link Long} values
- * that represent the latest (rolling) count (i.e., number of records) for each key within a window.
- * @deprecated use {@link #windowedBy(Windows) windowedBy(windows)} followed by
- * {@link TimeWindowedKStream#count(Materialized) count(Materialized.as(queryableStoreName))}
- */
- @Deprecated
-
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
- * and "-changelog" is a fixed suffix.
- * Note that the internal store name may not be queriable through Interactive Queries.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param windows the specification of the aggregation {@link Windows}
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys and {@link Long} values
- * that represent the latest (rolling) count (i.e., number of records) for each key within a window
- * @deprecated use {@link #windowedBy(Windows) windowedBy(windows)} followed by {@link TimeWindowedKStream#count() count()}
- */
- @Deprecated
-
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys and {@link Long} values
- * that represent the latest (rolling) count (i.e., number of records) for each key within a window
- * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
- * {@link SessionWindowedKStream#count() count()}
- */
- @Deprecated
- KTable
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
* The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
* aggregate and the record's value.
@@ -438,13 +134,16 @@ KTable
* For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
* The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
* user-specified in {@link StreamsConfig} via parameter
* {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
* and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * Note that the internal store name may not be queriable through Interactive Queries.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
* @return a {@link KTable} that contains "update" records with unmodified keys, and values that represent the
@@ -452,13 +151,14 @@ KTable
* The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
@@ -474,7 +174,8 @@ KTable
* If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
* value as-is.
- * Thus, {@code reduce(Reducer, String)} can be used to compute aggregate functions like sum, min, or max.
+ * Thus, {@code reduce(Reducer, Materialized)} can be used to compute aggregate functions like sum, min, or
+ * max.
*
* Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
* the same key.
@@ -484,61 +185,51 @@ KTable
* To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
+ * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
*
* For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
+ * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
* user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
+ * and "-changelog" is a fixed suffix.
+ * Note that the internal store name may not be queriable through Interactive Queries.
*
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param queryableStoreName the name of the underlying {@link KTable} state store; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} then this will be equivalent to {@link KGroupedStream#reduce(Reducer)} ()}.
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
+ *
+ * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
+ * @param materialized an instance of {@link Materialized} used to materialize a state store. Cannot be {@code null}.
* @return a {@link KTable} that contains "update" records with unmodified keys, and values that represent the
* latest (rolling) aggregate for each key
- * @deprecated use {@link #reduce(Reducer, Materialized) reduce(reducer, Materialized.as(queryableStoreName))}
*/
- @Deprecated
KTable
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate (first argument) and the record's value (second argument):
- *
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, org.apache.kafka.streams.processor.StateStoreSupplier)} can be used to compute
- * aggregate functions like sum, min, or max.
+ * The default value serde from config will be used for serializing the result.
+ * If a different serde is required then you should use {@link #aggregate(Initializer, Aggregator, Materialized)}.
*
* Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
* the same key.
@@ -546,440 +237,33 @@ KTable
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate (first argument) and the record's value (second argument):
- *
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, Materialized)} can be used to compute aggregate functions like sum, min, or
- * max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- *
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate (first argument) and the record's value (second argument):
- *
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, Windows, String)} can be used to compute aggregate functions like sum, min, or max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param reducer a {@link Reducer} that computes a new aggregate result
- * @param windows the specification of the aggregation {@link Windows}
- * @param queryableStoreName the name of the state store created from this operation; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} then this will be equivalent to {@link KGroupedStream#reduce(Reducer, Windows)}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
- * the latest (rolling) aggregate for each key within a window
- * @deprecated use {@link #windowedBy(Windows) windowedBy(windows)} followed by
- * {@link TimeWindowedKStream#reduce(Reducer, Materialized) reduce(reducer, Materialized.as(queryableStoreName))}
- */
- @Deprecated
-
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate and the record's value.
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, Windows, String)} can be used to compute aggregate functions like sum, min, or max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
- * and "-changelog" is a fixed suffix.
- * Note that the internal store name may not be queriable through Interactive Queries.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param reducer a {@link Reducer} that computes a new aggregate result
- * @param windows the specification of the aggregation {@link Windows}
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
- * the latest (rolling) aggregate for each key within a window
- * @deprecated use {@link #windowedBy(Windows) windowedBy(windows)} followed by
- * {@link TimeWindowedKStream#reduce(Reducer) reduce(reducer)}
- */
- @Deprecated
-
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate (first argument) and the record's value (second argument):
- *
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, Windows, org.apache.kafka.streams.processor.StateStoreSupplier)} can be used to
- * compute aggregate functions like sum, min, or max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate (first argument) and the record's value (second argument):
- *
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, SessionWindows, String)} can be used to compute aggregate functions like sum, min,
- * or max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param queryableStoreName the name of the state store created from this operation; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} then this will be equivalent to {@link KGroupedStream#reduce(Reducer, SessionWindows)}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
- * the latest (rolling) aggregate for each key within a window
- * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
- * {@link SessionWindowedKStream#reduce(Reducer, Materialized) reduce(reducer, Materialized.as(queryableStoreName))}
- */
- @Deprecated
- KTable
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate and the record's value.
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, SessionWindows, String)} can be used to compute aggregate functions like sum, min,
- * or max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
- * the latest (rolling) aggregate for each key within a window
- * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
- * {@link SessionWindowedKStream#reduce(Reducer) reduce(reducer)}
- */
- @Deprecated
- KTable
- * The specified {@link Reducer} is applied for each input record and computes a new aggregate using the current
- * aggregate (first argument) and the record's value (second argument):
- *
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, SessionWindows, org.apache.kafka.streams.processor.StateStoreSupplier)} can be used
- * to compute aggregate functions like sum, min, or max.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
- * the latest (rolling) aggregate for each key within a window
- * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
- * {@link SessionWindowedKStream#reduce(Reducer, Materialized) reduce(reducer, Materialized.as(KeyValueByteStoreSupplier))}
- */
- @Deprecated
- KTable
* The specified {@link Initializer} is applied once directly before the first input record is processed to
@@ -988,7 +272,7 @@ KTable
* Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
* the same key.
@@ -1001,86 +285,28 @@ KTable
* For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
+ * Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII
* alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
+ * The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is
* user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "storeName" is the
+ * provide store name defined in {@code Materialized}, and "-changelog" is a fixed suffix.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
* @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param queryableStoreName the name of the state store created from this operation; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} then this will be equivalent to {@link KGroupedStream#aggregate(Initializer, Aggregator, Serde)}.
- * @param
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Serde, String)} can be used to compute aggregate functions like
- * count (c.f. {@link #count()}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param materialized an instance of {@link Materialized} used to materialize a state store. Cannot be {@code null}.
+ * @param materialized an instance of {@link Materialized} used to materialize a state store. Cannot be {@code null}.
* @param
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator)} can be used to compute aggregate functions like
- * count (c.f. {@link #count()}).
- *
- * The default value serde from config will be used for serializing the result.
- * If a different serde is required then you should use {@link #aggregate(Initializer, Aggregator, Materialized)}.
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
- * and "-changelog" is a fixed suffix.
- * Note that the internal store name may not be queriable through Interactive Queries.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Serde, String)} can be used to compute aggregate functions like
- * count (c.f. {@link #count()}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
- * and "-changelog" is a fixed suffix.
- * Note that the internal store name may not be queriable through Interactive Queries.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, org.apache.kafka.streams.processor.StateStoreSupplier)} can be
- * used to compute aggregate functions like count (c.f. {@link #count()}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * The specified {@link Initializer} is applied once per window directly before the first input record is
- * processed to provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Windows, Serde, String)} can be used to compute aggregate
- * functions like count (c.f. {@link #count(Windows)}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * Therefore, the store name must be a valid Kafka topic name and cannot contain characters other than ASCII
- * alphanumerics, '.', '_' and '-'.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param windows the specification of the aggregation {@link Windows}
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * The specified {@link Initializer} is applied once per window directly before the first input record is
- * processed to provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Windows, Serde, String)} can be used to compute aggregate
- * functions like count (c.f. {@link #count(Windows)}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
- * and "-changelog" is a fixed suffix.
- * Note that the internal store name may not be queriable through Interactive Queries.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param windows the specification of the aggregation {@link Windows}
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * The specified {@link Initializer} is applied once per window directly before the first input record is
- * processed to provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Windows, org.apache.kafka.streams.processor.StateStoreSupplier)}
- * can be used to compute aggregate functions like count (c.f. {@link #count(Windows)}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local windowed {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * The specified {@link Initializer} is applied once per session directly before the first input record is
- * processed to provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Merger, SessionWindows, Serde, String)} can be used to compute
- * aggregate functions like count (c.f. {@link #count(SessionWindows)})
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link SessionStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- *
- * The specified {@link Initializer} is applied once per session directly before the first input record is
- * processed to provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code aggregate(Initializer, Aggregator, Merger, SessionWindows, Serde, String)} can be used to compute
- * aggregate functions like count (c.f. {@link #count(SessionWindows)})
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * @param initializer the instance of {@link Initializer}
- * @param aggregator the instance of {@link Aggregator}
- * @param sessionMerger the instance of {@link Merger}
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * The specified {@link Initializer} is applied once per session directly before the first input record is
- * processed to provide an initial intermediate aggregation result that is used to process the first record.
- * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
- * aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value.
- * Thus, {@code #aggregate(Initializer, Aggregator, Merger, SessionWindows, Serde, org.apache.kafka.streams.processor.StateStoreSupplier)}
- * can be used to compute aggregate functions like count (c.f. {@link #count(SessionWindows)}).
- *
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same window and key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link SessionStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}.
- * Use {@link org.apache.kafka.streams.processor.StateStoreSupplier#name()} to get the store name:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * The store name must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics,
- * '.', '_' and '-'.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
*
- * @param queryableStoreName the name of the underlying {@link KTable} state store; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} this is the equivalent of {@link KGroupedTable#count()}.
- * @return a {@link KTable} that contains "update" records with unmodified keys and {@link Long} values that
- * represent the latest (rolling) count (i.e., number of records) for each key
- * @deprecated use {@link #count(Materialized) count(Materialized.as(queryableStoreName))}
- */
- @Deprecated
- KTable
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
* For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
+ * Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII
+ * alphanumerics, '.', '_' and '-'.
+ * The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is
* user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * The store name must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics,
- * '.', '_' and '-'.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "storeName" is the
+ * provide store name defined in {@code Materialized}, and "-changelog" is a fixed suffix.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param materialized the instance of {@link Materialized} used to materialize the state store. Cannot be {@code null}
* @return a {@link KTable} that contains "update" records with unmodified keys and {@link Long} values that
@@ -147,128 +105,13 @@ public interface KGroupedTable
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
* @return a {@link KTable} that contains "update" records with unmodified keys and {@link Long} values that
* represent the latest (rolling) count (i.e., number of records) for each key
- * @deprecated use {@link #count(Materialized) count(Materialized.as(KeyValueByteStoreSupplier)}
*/
- @Deprecated
- KTable
- * Each update to the original {@link KTable} results in a two step update of the result {@link KTable}.
- * The specified {@link Reducer adder} is applied for each update record and computes a new aggregate using the
- * current aggregate (first argument) and the record's value (second argument) by adding the new record to the
- * aggregate.
- * The specified {@link Reducer subtractor} is applied for each "replaced" record of the original {@link KTable}
- * and computes a new aggregate using the current aggregate (first argument) and the record's value (second
- * argument) by "removing" the "replaced" record from the aggregate.
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, Reducer, String)} can be used to compute aggregate functions like sum.
- * For sum, the adder and subtractor would work as follows:
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * The store name must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics,
- * '.', '_' and '-'.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param adder a {@link Reducer} that adds a new value to the aggregate result
- * @param subtractor a {@link Reducer} that removed an old value from the aggregate result
- * @param queryableStoreName the name of the underlying {@link KTable} state store; valid characters are ASCII alphanumerics,
- * '.', '_' and '-'. If {@code null} this is the equivalent of {@link KGroupedTable#reduce(Reducer, Reducer)} ()}.
- * @return a {@link KTable} that contains "update" records with unmodified keys, and values that represent the
- * latest (rolling) aggregate for each key
- * @deprecated use {@link #reduce(Reducer, Reducer, Materialized) reduce(adder, subtractor, Materialized.as(queryableStoreName))}
- */
- @Deprecated
- KTable
* For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
+ * Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII
+ * alphanumerics, '.', '_' and '-'.
+ * The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is
* user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * The store name must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics,
- * '.', '_' and '-'.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "storeName" is the
+ * provide store name defined in {@code Materialized}, and "-changelog" is a fixed suffix.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param adder a {@link Reducer} that adds a new value to the aggregate result
* @param subtractor a {@link Reducer} that removed an old value from the aggregate result
@@ -386,7 +230,8 @@ KTable
- * Each update to the original {@link KTable} results in a two step update of the result {@link KTable}.
- * The specified {@link Reducer adder} is applied for each update record and computes a new aggregate using the
- * current aggregate (first argument) and the record's value (second argument) by adding the new record to the
- * aggregate.
- * The specified {@link Reducer subtractor} is applied for each "replaced" record of the original {@link KTable}
- * and computes a new aggregate using the current aggregate (first argument) and the record's value (second
- * argument) by "removing" the "replaced" record from the aggregate.
- * If there is no current aggregate the {@link Reducer} is not applied and the new aggregate will be the record's
- * value as-is.
- * Thus, {@code reduce(Reducer, Reducer, String)} can be used to compute aggregate functions like sum.
- * For sum, the adder and subtractor would work as follows:
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param adder a {@link Reducer} that adds a new value to the aggregate result
- * @param subtractor a {@link Reducer} that removed an old value from the aggregate result
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @return a {@link KTable} that contains "update" records with unmodified keys, and values that represent the
- * latest (rolling) aggregate for each key
- * @deprecated use {@link #reduce(Reducer, Reducer, Materialized) reduce(adder, subtractor, Materialized.as(KeyValueByteStoreSupplier))}
- */
- @Deprecated
- KTable
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * Each update to the original {@link KTable} results in a two step update of the result {@link KTable}.
- * The specified {@link Aggregator adder} is applied for each update record and computes a new aggregate using the
- * current aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value by adding the new record to the aggregate.
- * The specified {@link Aggregator subtractor} is applied for each "replaced" record of the original {@link KTable}
- * and computes a new aggregate using the current aggregate and the record's value by "removing" the "replaced"
- * record from the aggregate.
- * Thus, {@code aggregate(Initializer, Aggregator, Aggregator, String)} can be used to compute aggregate functions
- * like sum.
- * For sum, the initializer, adder, and subtractor would work as follows:
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer a {@link Initializer} that provides an initial aggregate result value
- * @param adder a {@link Aggregator} that adds a new record to the aggregate result
- * @param subtractor a {@link Aggregator} that removed an old record from the aggregate result
- * @param queryableStoreName the name of the underlying {@link KTable} state store.
- * If {@code null} this is the equivalent of {@link KGroupedTable#aggregate(Initializer, Aggregator, Aggregator)} ()}.
- * @param
* For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
+ * Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII
+ * alphanumerics, '.', '_' and '-'.
+ * The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is
* user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "storeName" is the
+ * provide store name defined in {@code Materialized}, and "-changelog" is a fixed suffix.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param initializer an {@link Initializer} that provides an initial aggregate result value
* @param adder an {@link Aggregator} that adds a new record to the aggregate result
@@ -640,7 +330,7 @@
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * Each update to the original {@link KTable} results in a two step update of the result {@link KTable}.
- * The specified {@link Aggregator adder} is applied for each update record and computes a new aggregate using the
- * current aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value by adding the new record to the aggregate.
- * The specified {@link Aggregator subtractor} is applied for each "replaced" record of the original {@link KTable}
- * and computes a new aggregate using the current aggregate and the record's value by "removing" the "replaced"
- * record from the aggregate.
- * Thus, {@code aggregate(Initializer, Aggregator, Aggregator, String)} can be used to compute aggregate functions
- * like sum.
- * For sum, the initializer, adder, and subtractor would work as follows:
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * The store name must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics,
- * '.', '_' and '-'.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer a {@link Initializer} that provides an initial aggregate result value
- * @param adder a {@link Aggregator} that adds a new record to the aggregate result
- * @param subtractor a {@link Aggregator} that removed an old record from the aggregate result
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param queryableStoreName the name of the underlying {@link KTable} state store; valid characters are ASCII
- * alphanumerics, '.', '_' and '-'. If {@code null} this is the equivalent of {@link KGroupedTable#aggregate(Initializer, Aggregator, Aggregator, Serde)} ()}.
- * @param
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * Each update to the original {@link KTable} results in a two step update of the result {@link KTable}.
- * The specified {@link Aggregator adder} is applied for each update record and computes a new aggregate using the
- * current aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value by adding the new record to the aggregate.
- * The specified {@link Aggregator subtractor} is applied for each "replaced" record of the original {@link KTable}
- * and computes a new aggregate using the current aggregate and the record's value by "removing" the "replaced"
- * record from the aggregate.
- * Thus, {@code aggregate(Initializer, Aggregator, Aggregator, String)} can be used to compute aggregate functions
- * like sum.
- * For sum, the initializer, adder, and subtractor would work as follows:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
- * and "-changelog" is a fixed suffix.
- * Note that the internal store name may not be queriable through Interactive Queries.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer a {@link Initializer} that provides an initial aggregate result value
- * @param adder a {@link Aggregator} that adds a new record to the aggregate result
- * @param subtractor a {@link Aggregator} that removed an old record from the aggregate result
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * The specified {@link Initializer} is applied once directly before the first input record is processed to
- * provide an initial intermediate aggregation result that is used to process the first record.
- * Each update to the original {@link KTable} results in a two step update of the result {@link KTable}.
- * The specified {@link Aggregator adder} is applied for each update record and computes a new aggregate using the
- * current aggregate (or for the very first record using the intermediate aggregation result provided via the
- * {@link Initializer}) and the record's value by adding the new record to the aggregate.
- * The specified {@link Aggregator subtractor} is applied for each "replaced" record of the original {@link KTable}
- * and computes a new aggregate using the current aggregate and the record's value by "removing" the "replaced"
- * record from the aggregate.
- * Thus, {@code aggregate(Initializer, Aggregator, Aggregator, String)} can be used to compute aggregate functions
- * like sum.
- * For sum, the initializer, adder, and subtractor would work as follows:
- *
- * To query the local {@link KeyValueStore} it must be obtained via
- * {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
- *
- * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-${queryableStoreName}-changelog", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "queryableStoreName" is the
- * provide {@code queryableStoreName}, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param initializer a {@link Initializer} that provides an initial aggregate result value
- * @param adder a {@link Aggregator} that adds a new record to the aggregate result
- * @param subtractor a {@link Aggregator} that removed an old record from the aggregate result
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @param
- * The default serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- * @deprecated use {@code print(Printed)}
- */
- @Deprecated
- void print();
-
- /**
- * Print the records of this stream to {@code System.out}.
- * This function will use the given name to label the key/value pairs printed to the console.
- *
- * The default serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param label the name used to label the key/value pairs printed to the console
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut()}
- */
- @Deprecated
- void print(final String label);
-
- /**
- * Print the records of this stream to {@code System.out}.
- * This function will use the generated name of the parent processor node to label the key/value pairs printed to
- * the console.
- *
- * The provided serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param keySerde key serde used to deserialize key if type is {@code byte[]},
- * @param valSerde value serde used to deserialize value if type is {@code byte[]},
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut().withKeyValueMapper(...)}
- */
- @Deprecated
- void print(final Serde
- * The provided serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param keySerde key serde used to deserialize key if type is {@code byte[]},
- * @param valSerde value serde used to deserialize value if type is {@code byte[]},
- * @param label the name used to label the key/value pairs printed to the console
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut().withLabel(label).withKeyValueMapper(...)}
- */
- @Deprecated
- void print(final Serde
- * The default serde will be use to deserialize key or value if type is {@code byte[]}.
- * The user provided {@link KeyValueMapper} which customizes output is used to print with {@code System.out}
- *
- * The example below shows the way to customize output data.
- *
- * The KeyValueMapper's mapped value type must be {@code String}.
- *
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut().withKeyValueMapper(mapper)}
- */
- @Deprecated
- void print(final KeyValueMapper mapper);
-
- /**
- * Print the customized output with {@code System.out}.
- *
- * The default serde will be used to deserialize key or value if type is {@code byte[]}.
- * The user provided {@link KeyValueMapper} which customizes output is used to print with {@code System.out}
- *
- * The example below shows the way to customize output data.
- *
- * The KeyValueMapper's mapped value type must be {@code String}.
- *
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @param label The given name which labels output will be printed.
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut().withLabel(label).withKeyValueMapper(mapper)}
- */
- @Deprecated
- void print(final KeyValueMapper mapper, final String label);
-
- /**
- * Print the customized output with {@code System.out}.
- *
- * The user provided {@link KeyValueMapper} which customizes output is used to print with {@code System.out}
- * The provided serde will be use to deserialize key or value if type is {@code byte[]}.
- *
- * The example below shows the way to customize output data.
- *
- * The provided KeyValueMapper's mapped value type must be {@code String}.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @param keySerde a {@link Serde} used to deserialize key if type is {@code byte[]}.
- * @param valSerde a {@link Serde} used to deserialize value if type is {@code byte[]}.
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut().withKeyValueMapper(mapper)}
- */
- @Deprecated
- void print(final KeyValueMapper mapper, final Serde
- * The user provided {@link KeyValueMapper} which customizes output is used to print with {@code System.out}.
- * The provided serde will be use to deserialize key or value if type is {@code byte[]}.
- *
- * The example below shows the way to customize output data.
- *
- * The provided KeyValueMapper's mapped value type must be {@code String}.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @param keySerde a {@link Serde} used to deserialize key if type is {@code byte[]}.
- * @param valSerde a {@link Serde} used to deserialize value if type is {@code byte[]}.
- * @param label The given name which labels output will be printed.
- * @deprecated use {@link #print(Printed) print(Printed.toSysOut().withLabel(label).withKeyValueMapper(mapper)}
- */
- @Deprecated
- void print(final KeyValueMapper mapper, final Serde
- * There is no ordering guarantee between records from this {@code KStream} and records from
- * the provided {@code KStream} in the merged stream.
- * Relative order is preserved within each input stream though (ie, records within one input
- * stream are processed in order).
- *
- * @param stream a stream which is to be merged into this stream
- * @return a merged stream containing all records from this and the provided {@code KStream}
- */
- KStream
- * The default serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param filePath name of the file to write to
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath)}
- */
- @Deprecated
- void writeAsText(final String filePath);
-
- /**
- * Write the records of this stream to a file at the given path.
- * This function will use the given name to label the key/value printed to the file.
- *
- * The default serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param filePath name of the file to write to
- * @param label the name used to label the key/value pairs written to the file
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withLabel(label)}
- */
- @Deprecated
- void writeAsText(final String filePath,
- final String label);
-
- /**
- * Write the records of this stream to a file at the given path.
- * This function will use the generated name of the parent processor node to label the key/value pairs printed to
- * the file.
- *
- * The provided serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
- * {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param filePath name of the file to write to
- * @param keySerde key serde used to deserialize key if type is {@code byte[]},
- * @param valSerde value serde used to deserialize value if type is {@code byte[]},
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withKeyValueMapper(...)}
- */
- @Deprecated
- void writeAsText(final String filePath,
- final Serde
- * The provided serde will be used to deserialize the key or value in case the type is {@code byte[]}
- * before calling {@code toString()} on the deserialized object.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param filePath name of the file to write to
- * @param label the name used to label the key/value pairs written to the file
- * @param keySerde key serde used to deserialize key if type is {@code byte[]},
- * @param valSerde value serde used deserialize value if type is {@code byte[]},
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(...)}
- */
- @Deprecated
- void writeAsText(final String filePath,
- final String label,
- final Serde
- * The user provided {@link KeyValueMapper} which customizes output is used to write to file.
- * This function will use default name of stream to label records.
- *
- * The default key and value serde will used to deserialize {@code byte[]} records before calling {@code toString()}.
- *
- * The example below shows the way to customize output data.
- *
- * The KeyValueMapper's mapped value type must be {@code String}.
- *
- * @param filePath path of the file to write to.
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withKeyValueMapper(mapper)}
- */
- @Deprecated
- void writeAsText(final String filePath, final KeyValueMapper mapper);
-
- /**
- * Write the customised output to a given file path.
- *
- * The user provided {@link KeyValueMapper} which customizes output is used to write to file.
- * This function will use given name of stream to label records.
- *
- * The default key and value serde will used to deserialize {@code byte[]} records before calling {@code toString()}.
- *
- * The example below shows the way to customize output data.
- *
- * The KeyValueMapper's mapped value type must be {@code String}.
- *
- * @param filePath path of the file to write to.
- * @param label the name used to label records written to file.
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(mapper)}
- */
- @Deprecated
- void writeAsText(final String filePath, final String label, final KeyValueMapper mapper);
-
- /**
- * Write the customised output to a given file path.
- *
- * The user provided {@link KeyValueMapper} which customizes output is used to write to file.
- * This function will use default name of stream to label records.
- *
- * The given key and value serde will be used to deserialize {@code byte[]} records before calling {@code toString()}.
- *
- * The example below shows the way to customize output data.
- *
- * The KeyValueMapper's mapped value type must be {@code String}.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param filePath path of the file to write to.
- * @param keySerde key serde used to deserialize key if type is {@code byte[]}.
- * @param valSerde value serde used to deserialize value if type is {@code byte[]}.
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withKeyValueMapper(mapper)}
- */
- @Deprecated
- void writeAsText(final String filePath, final Serde
- * The user provided {@link KeyValueMapper} which customizes output is used to write to file.
- * This function will use given name of stream to label records.
- *
- * The given key and value serde will be used to deserialize {@code byte[]} records before calling {@code toString()}.
- *
- * The example below shows the way to customize output data.
- *
- * The KeyValueMapper's mapped value type must be {@code String}.
- *
- * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
- * {@link Integer} etc. to get meaningful information.
- *
- * @param filePath path of the file to write to.
- * @param label the name used to label records written to file.
- * @param keySerde key serde used to deserialize key if type is {@code byte[]}.
- * @param valSerde value serde used to deserialize value if type is {@code byte[]}.
- * @param mapper a {@link KeyValueMapper} that computes output type {@code String}.
- * @deprecated use {@link #print(Printed) print(Printed.toFile(filePath).withLabel(label).withKeyValueMapper(mapper)}
- */
- @Deprecated
- void writeAsText(final String filePath, final String label, final Serde
- * This is equivalent to calling {@link #to(String) #to(someTopicName)} and
- * {@link StreamsBuilder#stream(String) StreamsBuilder#stream(someTopicName)}.
+ * There is no ordering guarantee between records from this {@code KStream} and records from
+ * the provided {@code KStream} in the merged stream.
+ * Relative order is preserved within each input stream though (ie, records within one input
+ * stream are processed in order).
*
- * @param topic the topic name
- * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
+ * @param stream a stream which is to be merged into this stream
+ * @return a merged stream containing all records from this and the provided {@code KStream}
*/
- KStream
- * This is equivalent to calling {@link #to(StreamPartitioner, String) #to(StreamPartitioner, someTopicName)} and
+ * This is equivalent to calling {@link #to(String) #to(someTopicName)} and
* {@link StreamsBuilder#stream(String) StreamsBuilder#stream(someTopicName)}.
*
- * @param partitioner the function used to determine how records are distributed among partitions of the topic,
- * if not specified producer's {@link DefaultPartitioner} will be used
- * @param topic the topic name
- * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
- * @deprecated use {@link #through(String, Produced) through(topic, Produced.withStreamPartitioner(partitioner))}
- */
- @Deprecated
- KStream
- * If {@code keySerde} provides a {@link WindowedSerializer} for the key {@link WindowedStreamPartitioner} is
- * used—otherwise producer's {@link DefaultPartitioner} is used.
- *
- * This is equivalent to calling {@link #to(Serde, Serde, String) #to(keySerde, valSerde, someTopicName)} and
- * {@link KStreamBuilder#stream(Serde, Serde, String...) KStreamBuilder#stream(keySerde, valSerde, someTopicName)}.
- *
- * @param keySerde key serde used to send key-value pairs,
- * if not specified the default key serde defined in the configuration will be used
- * @param valSerde value serde used to send key-value pairs,
- * if not specified the default value serde defined in the configuration will be used
- * @param topic the topic name
- * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
- * @deprecated use {@link #through(String, Produced) through(topic, Produced.with(keySerde, valSerde))}
- */
- @Deprecated
- KStream
- * This is equivalent to calling {@link #to(Serde, Serde, StreamPartitioner, String) #to(keySerde, valSerde,
- * StreamPartitioner, someTopicName)} and {@link KStreamBuilder#stream(Serde, Serde, String...)
- * KStreamBuilder#stream(keySerde, valSerde, someTopicName)}.
- *
- * @param keySerde key serde used to send key-value pairs,
- * if not specified the default key serde defined in the configuration will be used
- * @param valSerde value serde used to send key-value pairs,
- * if not specified the default value serde defined in the configuration will be used
- * @param partitioner the function used to determine how records are distributed among partitions of the topic,
- * if not specified and {@code keySerde} provides a {@link WindowedSerializer} for the key
- * {@link WindowedStreamPartitioner} will be used—otherwise {@link DefaultPartitioner} will
- * be used
- * @param topic the topic name
+ * @param topic the topic name
* @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
- * @deprecated use {@link #through(String, Produced) through(topic, Produced.with(keySerde, valSerde, partitioner))}
*/
- @Deprecated
- KStream
* For this case, all data of this stream will be redistributed through the repartitioning topic by writing all
* records to it, and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned
@@ -1276,7 +781,9 @@ void process(final ProcessorSupplier processorSupplier,
* This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in
* {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "XXX" is
* an internally generated name, and "-repartition" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
+ *
*
* For this case, all data of this stream will be redistributed through the repartitioning topic by writing all
* records to it, and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned
@@ -1287,36 +794,6 @@ void process(final ProcessorSupplier processorSupplier,
*/
KGroupedStream
- * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)},
- * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)}, or
- * {@link #transform(TransformerSupplier, String...)}), and no data redistribution happened afterwards (e.g., via
- * {@link #through(String)}) an internal repartitioning topic will be created in Kafka.
- * This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in
- * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "XXX" is
- * an internally generated name, and "-repartition" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * For this case, all data of this stream will be redistributed through the repartitioning topic by writing all
- * records to it, and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned
- * correctly on its key.
- *
- * @param keySerde key serdes for materializing this stream,
- * if not specified the default serdes defined in the configs will be used
- * @param valSerde value serdes for materializing this stream,
- * if not specified the default serdes defined in the configs will be used
- * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KStream}
- * @deprecated use {@link #groupByKey(Serialized) groupByKey(Serialized.with(keySerde, valSerde))}
- */
- @Deprecated
- KGroupedStream
* All data of this stream will be redistributed through the repartitioning topic by writing all records to it,
* and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned on the new key.
@@ -1355,7 +834,9 @@ KGroupedStream
* All data of this stream will be redistributed through the repartitioning topic by writing all records to it,
* and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned on the new key.
@@ -1369,40 +850,6 @@ KGroupedStream
- * Because a new key is selected, an internal repartitioning topic will be created in Kafka.
- * This topic will be named "${applicationId}-XXX-repartition", where "applicationId" is user-specified in
- * {@link StreamsConfig StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "XXX" is an internally generated name, and
- * "-repartition" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * All data of this stream will be redistributed through the repartitioning topic by writing all records to it,
- * and rereading all records from it, such that the resulting {@link KGroupedStream} is partitioned on the new key.
- *
- * This is equivalent to calling {@link #selectKey(KeyValueMapper)} followed by {@link #groupByKey(Serde, Serde)}.
- *
- * @param selector a {@link KeyValueMapper} that computes a new key for grouping
- * @param keySerde key serdes for materializing this stream,
- * if not specified the default serdes defined in the configs will be used
- * @param valSerde value serdes for materializing this stream,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
- * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
- * in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "storeName" is an
- * internally generated name, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
*
- * @param otherStream the {@code KStream} to be joined with this stream
- * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
- * @param windows the specification of the {@link JoinWindows}
- * @param
- * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
- * a value (with arbitrary type) for the result record.
- * The key of the result record is the same as for both joining input records.
- * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
- * output record will be added to the resulting {@code KStream}.
- *
- * Example (assuming all input records belong to the correct windows):
- *
* Repartitioning can happen for one or both of the joining {@code KStream}s.
* For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
@@ -1539,173 +910,8 @@
- * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
- * a value (with arbitrary type) for the result record.
- * The key of the result record is the same as for both joining input records.
- * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
- * output record will be added to the resulting {@code KStream}.
- *
- * Example (assuming all input records belong to the correct windows):
- *
- * Repartitioning can happen for one or both of the joining {@code KStream}s.
- * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
- * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
- * correctly on its key.
- *
- * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
- * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
- * in {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
- * "storeName" is an internally generated name, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- * @param otherStream the {@code KStream} to be joined with this stream
- * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
- * @param windows the specification of the {@link JoinWindows}
- * @param keySerde key serdes for materializing both streams,
- * if not specified the default serdes defined in the configs will be used
- * @param thisValueSerde value serdes for materializing this stream,
- * if not specified the default serdes defined in the configs will be used
- * @param otherValueSerde value serdes for materializing the other stream,
- * if not specified the default serdes defined in the configs will be used
- * @param
- * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
- * a value (with arbitrary type) for the result record.
- * The key of the result record is the same as for both joining input records.
- * Furthermore, for each input record of this {@code KStream} that does not satisfy the join predicate the provided
- * {@link ValueJoiner} will be called with a {@code null} value for the other stream.
- * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
- * output record will be added to the resulting {@code KStream}.
- *
- * Example (assuming all input records belong to the correct windows):
- *
- * Repartitioning can happen for one or both of the joining {@code KStream}s.
- * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
- * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
- * correctly on its key.
- *
- * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
- * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
- * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
- * in {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
- * "storeName" is an internally generated name, and "-changelog" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param otherStream the {@code KStream} to be joined with this stream
* @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
@@ -1713,20 +919,17 @@
@@ -1749,7 +950,7 @@
* Repartitioning can happen for one or both of the joining {@code KStream}s.
* For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
@@ -1783,9 +983,11 @@
* Repartitioning can happen for one or both of the joining {@code KStream}s.
* For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
@@ -1865,43 +1066,31 @@
@@ -1929,12 +1118,12 @@
* Repartitioning can happen for one or both of the joining {@code KStream}s.
* For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
@@ -1960,22 +1148,26 @@ {@code
* KafkaStreams streams = ... // counting words
- * String queryableStoreName = storeSupplier.name();
+ * String queryableStoreName = "storeName"; // the store name should be the name of the store as defined by the Materialized instance
* ReadOnlyKeyValueStore
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * KafkaStreams streams = ... // counting words
- * String queryableStoreName = "count-store"; // the queryableStoreName should be the name of the store as defined by the Materialized instance
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
+ * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
+ * Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII
+ * alphanumerics, '.', '_' and '-'.
+ * The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is
+ * user-specified in {@link StreamsConfig} via parameter
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "storeName" is the
+ * provide store name defined in {@code Materialized}, and "-changelog" is a fixed suffix.
+ *
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
*
* @param materialized an instance of {@link Materialized} used to materialize a state store. Cannot be {@code null}.
* Note: the valueSerde will be automatically set to {@link org.apache.kafka.common.serialization.Serdes#Long() Serdes#Long()}
@@ -183,248 +115,12 @@ public interface KGroupedStream{@code
- * KafkaStreams streams = ... // counting words
- * ReadOnlyWindowStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * KafkaStreams streams = ... // counting words
- * String queryableStoreName = storeSupplier.name();
- * ReadOnlyWindowStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param windows the specification of the aggregation {@link Windows}
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys and {@link Long} values
- * that represent the latest (rolling) count (i.e., number of records) for each key within a window
- * @deprecated use {@link #windowedBy(Windows) windowedBy(windows)} followed by
- * {@link TimeWindowedKStream#count(Materialized) count(Materialized.as(KeyValueByteStoreSupplier))}
- */
- @Deprecated
- {@code
- * KafkaStreams streams = ... // compute sum
- * ReadOnlySessionStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param queryableStoreName the name of the state store created from this operation; valid characters are ASCII
- * alphanumerics, '.', '_' and '-. If {@code null} then this will be equivalent to {@link KGroupedStream#count(SessionWindows)}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys and {@link Long} values
- * that represent the latest (rolling) count (i.e., number of records) for each key within a window
- * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
- * {@link SessionWindowedKStream#count(Materialized) count(Materialized.as(queryableStoreName))}
- */
- @Deprecated
- KTable{@code
- * KafkaStreams streams = ... // compute sum
- * Sting queryableStoreName = storeSupplier.name();
- * ReadOnlySessionStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys and {@link Long} values
- * that represent the latest (rolling) count (i.e., number of records) for each key within a window
- * @deprecated use {@link #windowedBy(SessionWindows) windowedBy(sessionWindows)} followed by
- * {@link SessionWindowedKStream#count(Materialized) count(Materialized.as(KeyValueByteStoreSupplier))}
- */
- @Deprecated
- KTable{@code
* KafkaStreams streams = ... // compute sum
+ * String queryableStoreName = "storeName" // the store name should be the name of the store as defined by the Materialized instance
* ReadOnlyKeyValueStore
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
+ *
* {@code
- * // At the example of a Reducer
+ * The specified {@link Initializer} is applied once directly before the first input record is processed to
+ * provide an initial intermediate aggregation result that is used to process the first record.
+ * The specified {@link Aggregator} is applied for each input record and computes a new aggregate using the current
+ * aggregate (or for the very first record using the intermediate aggregation result provided via the
+ * {@link Initializer}) and the record's value.
+ * Thus, {@code aggregate(Initializer, Aggregator)} can be used to compute aggregate functions like
+ * count (c.f. {@link #count()}).
* {@code
- * KafkaStreams streams = ... // compute sum
- * String queryableStoreName = storeSupplier.name();
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
+ * For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka.
+ * The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is
+ * user-specified in {@link StreamsConfig} via parameter
+ * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "internalStoreName" is an internal name
+ * and "-changelog" is a fixed suffix.
+ * Note that the internal store name may not be queriable through Interactive Queries.
*
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
+ * You can retrieve all generated internal topic names via {@link Topology#describe()}.
+ *
+ * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
+ * @param aggregator an {@link Aggregator} that computes a new aggregate result
+ * @param {@code
- * // At the example of a Reducer
- * {@code
- * KafkaStreams streams = ... // compute sum
- * String queryableStoreName = "storeName" // the queryableStoreName should be the name of the store as defined by the Materialized instance
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param materialized an instance of {@link Materialized} used to materialize a state store. Cannot be {@code null}.
- * @return a {@link KTable} that contains "update" records with unmodified keys, and values that represent the
- * latest (rolling) aggregate for each key
- */
- KTable{@code
- * // At the example of a Reducer
- * {@code
- * KafkaStreams streams = ... // compute sum
- * ReadOnlyWindowStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * // At the example of a Reducer
- * {@code
- * KafkaStreams streams = ... // compute sum
- * Sting queryableStoreName = storeSupplier.name();
- * ReadOnlyWindowStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param reducer a {@link Reducer} that computes a new aggregate result. Cannot be {@code null}.
- * @param windows the specification of the aggregation {@link Windows}
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @return a windowed {@link KTable} that contains "update" records with unmodified keys, and values that represent
- * the latest (rolling) aggregate for each key within a window
- * @deprecated use {@link #windowedBy(Windows) windowedBy(windows)} followed by
- * {@link TimeWindowedKStream#reduce(Reducer, Materialized) reduce(reducer, Materialized.as(KeyValueByteStoreSupplier))}
- */
- @Deprecated
- {@code
- * // At the example of a Reducer
- * {@code
- * KafkaStreams streams = ... // compute sum
- * ReadOnlySessionStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * // At the example of a Reducer
- * {@code
- * KafkaStreams streams = ... // compute sum
- * Sting queryableStoreName = storeSupplier.name();
- * ReadOnlySessionStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
* KafkaStreams streams = ... // some aggregation on value type double
+ * String queryableStoreName = "storeName" // the store name should be the name of the store as defined by the Materialized instance
* ReadOnlyKeyValueStore
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
+ *
* {@code
- * KafkaStreams streams = ... // some aggregation on value type double
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * KafkaStreams streams = ... // some aggregation on value type double
- * Sting queryableStoreName = storeSupplier.name();
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @param {@code
- * KafkaStreams streams = ... // some windowed aggregation on value type double
- * ReadOnlyWindowStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * KafkaStreams streams = ... // some windowed aggregation on value type Long
- * Sting queryableStoreName = storeSupplier.name();
- * ReadOnlyWindowStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param initializer an {@link Initializer} that computes an initial intermediate aggregation result
- * @param aggregator an {@link Aggregator} that computes a new aggregate result
- * @param windows the specification of the aggregation {@link Windows}
- * @param {@code
- * KafkaStreams streams = ... // some windowed aggregation on value type double
- * ReadOnlySessionStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- * @param initializer the instance of {@link Initializer}
- * @param aggregator the instance of {@link Aggregator}
- * @param sessionMerger the instance of {@link Merger}
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param {@code
- * KafkaStreams streams = ... // some windowed aggregation on value type double
- * Sting queryableStoreName = storeSupplier.name();
- * ReadOnlySessionStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- *
- *
- * @param initializer the instance of {@link Initializer}
- * @param aggregator the instance of {@link Aggregator}
- * @param sessionMerger the instance of {@link Merger}
- * @param sessionWindows the specification of the aggregation {@link SessionWindows}
- * @param aggValueSerde aggregate value serdes for materializing the aggregated table,
- * if not specified the default serdes defined in the configs will be used
- * @param storeSupplier user defined state store supplier. Cannot be {@code null}.
- * @param {@code
- * KafkaStreams streams = ... // counting words
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
* {@code
- * KafkaStreams streams = ... // counting words
- * String queryableStoreName = storeSupplier.name();
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * public class SumAdder implements Reducer
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * {@code
- * KafkaStreams streams = ... // counting words
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * public class SumAdder implements Reducer
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * {@code
- * KafkaStreams streams = ... // counting words
- * String queryableStoreName = storeSupplier.name();
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * // in this example, LongSerde.class must be set as default value serde in StreamsConfig
- * public class SumInitializer implements Initializer
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * {@code
- * KafkaStreams streams = ... // counting words
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * public class SumInitializer implements Initializer
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * {@code
- * KafkaStreams streams = ... // counting words
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * public class SumInitializer implements Initializer
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * {@code
- * public class SumInitializer implements Initializer
- * Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to
- * the same key.
- * The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of
- * parallel running Kafka Streams instances, and the {@link StreamsConfig configuration} parameters for
- * {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG cache size}, and
- * {@link StreamsConfig#COMMIT_INTERVAL_MS_CONFIG commit intervall}.
- * {@code
- * KafkaStreams streams = ... // counting words
- * String queryableStoreName = storeSupplier.name();
- * ReadOnlyKeyValueStore
- * For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
- * query the value of the key on a parallel running instance of your Kafka Streams application.
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- * {@code
- * final KeyValueMapper
- *
- *
- * Both input streams (or to be more precise, their underlying source topics) need to have the same number of
- * partitions.
- * If this is not the case, you would need to call {@link #through(String)} (for one input stream) before doing the
- * join, using a pre-created topic with the "correct" number of partitions.
- * Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
- * If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
- * internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
- * The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "XXX" is an internally generated name, and
- * "-repartition" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
*
- *
- * this
- * other
- * result
- *
- *
- * <K1:A>
- *
- *
- *
- *
- * <K2:B>
- * <K2:b>
- * <K2:ValueJoiner(B,b)>
- *
- *
- *
- * <K3:c>
- *
- *
- *
- * Both input streams (or to be more precise, their underlying source topics) need to have the same number of
- * partitions.
- * If this is not the case, you would need to call {@link #through(String)} (for one input stream) before doing the
- * join, using a pre-created topic with the "correct" number of partitions.
- * Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
- * If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
- * internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
- * The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "XXX" is an internally generated name, and
- * "-repartition" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- *
- * this
- * other
- * result
- *
- *
- * <K1:A>
- *
- *
- *
- *
- * <K2:B>
- * <K2:b>
- * <K2:ValueJoiner(B,b)>
- *
- *
- *
- * <K3:c>
- *
- *
- *
- * Both input streams (or to be more precise, their underlying source topics) need to have the same number of
- * partitions.
- * If this is not the case, you would need to call {@link #through(String)} (for one input stream) before doing the
- * join, using a pre-created topic with the "correct" number of partitions.
- * Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
- * If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
- * internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
- * The repartitioning topic will be named "${applicationId}-XXX-repartition", where "applicationId" is
- * user-specified in {@link StreamsConfig} via parameter
- * {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG}, "XXX" is an internally generated name, and
- * "-repartition" is a fixed suffix.
- * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
- *
- *
- * this
- * other
- * result
- *
- *
- * <K1:A>
- *
- * <K1:ValueJoiner(A,null)>
- *
- *
- * <K2:B>
- * <K2:b>
- * <K2:ValueJoiner(B,b)>
- *
- *
- *
- * <K3:c>
- *
- *
*
* <K1:A>
*
- * <K1:ValueJoiner(A,null)>
+ *
*
* <K2:B>
@@ -1770,10 +971,9 @@
*
* <K2:B>
* <K2:b>
- * <K2:ValueJoiner(null,b)>
+ *
<K2:ValueJoiner(B,b)><K2:ValueJoiner(B,b)>
*
*
*
* Both input streams (or to be more precise, their underlying source topics) need to have the same number of
@@ -1948,7 +1137,6 @@
* <K3:c>
- * <K3:ValueJoiner(null,c)>
+ *
*