...

JIRA:

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-4125

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-3907

Jira

server	ASF JIRA
columns	key,summary,type,created,updated,due,assignee,reporter,priority,status,resolution
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-4218

Jira

server	ASF JIRA
columns	key,summary,type,created,updated,due,assignee,reporter,priority,status,resolution
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-4726

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

This KIP combines KIP-149 and provides a hybrd hybrid solution to rich functions in Streams and accessing read-only keys within XXXValues operatorsValueJoiner, ValueTransformer, ValueMapper interfaces.

Rich functions are one of the essential parts of stream processing. There are several use-cases where users cannot express their business logic with current un-rich methods. For example:

...

having access to RecordContext within an operator
having access to a read-only key for XXXValues operatorsValueJoiner, ValueTransformer, ValueMapper interfaces

Rich Interfaces

Code Block

language	java

 
public interface RichValueMapper<KRichInitializer<V, V,K> VR> {
    VRV apply(final K key, final V value, final RecordContext recordContext);
}

public interface RichValueJoiner<KRichValueMapper<V, V1VR, V2, VR>K> {
    VR apply(final V value, final K key, final RecordContext recordContext);
}

public interface RichValueJoiner<V1, V2, VR, K> {
    VR apply(final V1 value1, final V2 value2, final K key, final RecordContext recordContext);
}

public interface RichKeyValueMapper<K, V, VR> {
    VR apply(final K key, final V value, final RecordContext recordContext);
}

public interface RichReducer<KRichReducer<V, V>K> {
    V apply(final KV keyvalue1, final V value1value2, final VK value2key, final RecordContext recordContext);
}

public interface RichAggregator<K, V, VA> {
    VA apply(final K key, final V value, final VA aggregate, final RecordContext recordContext);
}
 
public interface RichForeachAction<K, V> {
    void apply(final K key, final V value, final RecordContext recordContext);
}

public interface RichPredicate<K, V> {
    boolean test(final K key, final V value, final RecordContext recordContext);
}

 
public interface RichMerger<K, V> {
	V apply(final K aggKey, final V aggOne, final V aggTwo, final RecordContext recordContext);
}

 
 
public interface RichValueTransformer<KRichValueTransformer<V, VVR, VR>K> {
	void init(final ProcessorContext context);

	VR transform(final V value, final K key);

	@Deprecated
	VR punctuate(final K key, final long timestamp);

	void close();
void close();
}


 
public interface RichValueTransformerSupplier<KRichValueTransformerSupplier<V, VVR, VR>K> {

    RichValueTransformer<KRichValueTransformer<V, VVR, VR>K> get();
}

Public Interfaces

...

Code Block

language	java

KStream<K, V> filter(RichPredicate<? super K, ? super V> predicate);
KStream<K, V> filterNot(RichPredicate<? super K, ? super V> predicate);
<KR> KStream<KR, V> selectKey(RichKeyValueMapper<? super K, ? super V, ? extends KR> mapper);
<KR, VR> KStream<KR, VR> map(RichKeyValueMapper<? super K, ? super V, ? extends KeyValue<? extends KR, ? extends VR>> mapper);
<VR> KStream<K, VR> mapValues(RichValueMapper<? super KV, ? superextends VVR, ? extendssuper VR>K> mapper);
<KR, VR> KStream<KR, VR> flatMap(final RichKeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends KR, ? extends VR>>> mapper);
 
<VR> KStream<K, VR> flatMapValues(final RichValueMapper<? super KV, ? superextends V, Iterable<? extends Iterable<VR>, ? extendssuper VR>>K> mapper);



void foreach(final RichForeachAction<? super K, ? super V> action);
KStream<K, V> peek(final RichForeachAction<? super K, ? super V> action);
KStream<K, V>[] branch(final RichPredicate<? super K, ? super V>... predicates);
 

<VR> KStream<K, VR> transformValues(final RichValueTransformerSupplier<? super KV, ? superextends VVR, ? extendssuper VR>K> valueTransformerSupplier,
                                    final String... stateStoreNames);
<KR> KGroupedStream<KR, V> groupBy(final RichKeyValueMapper<? super K, ? super V, KR> selector);
<KR> KGroupedStream<KR, V> groupBy(final RichKeyValueMapper<? super K, ? super V, KR> selector,
                                   final Serialized<KR, V> serialized);
<KR> KGroupedStream<KR
 

<VO, VR> KStream<K, V>VR> groupByjoin(final RichKeyValueMapper<?KStream<K, superVO> KotherStream,
 ? super V, KR> selector,
                        final RichValueJoiner<? super V, ? super VO, ? extends VR, ? finalsuper Serde<KR>K> keySerdejoiner,
                                   final Serde<V>JoinWindows valSerdewindows);
 

<VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
                             final RichValueJoiner<? super KV, ? super VVO, ? superextends VOVR, ? extendssuper VR>K> joiner,
                             final JoinWindows windows);,
<VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
                        final Joined<K, V, VO> joined);
 

<VO, VR> KStream<K, VR> leftJoin(final RichValueJoiner<?KStream<K, superVO> KotherStream, ? super V, ? super VO, ? extends VR> joiner,

                                 final RichValueJoiner<? super V, ? super VO, ? extends VR, ? finalsuper JoinWindowsK> windowsjoiner,
                             final   Joined<K, V,final VO>JoinWindows joinedwindows);
 

<VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
                                 final RichValueJoiner<? super KV, ? super VVO, ? superextends VOVR, ? extendssuper VR>K> joiner,
                                 final JoinWindows windows,
                                 final Joined<K, V, VO> joined);
 

<VO, VR> KStream<K, VR> leftJoinouterJoin(final KStream<K, VO> otherStream,
                                  final RichValueJoiner<? super KV, ? super VVO, ? superextends VOVR, ? extendssuper VR>K> joiner,
                                  final JoinWindows windows,
      );
<VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
                           final Joined<K, V, VO> joined);
 

<VO, VR> KStream<K, VR> outerJoin(final KStream<K,RichValueJoiner<? VO>super otherStreamV,
 ? super VO, ? extends VR, ? super K> joiner,
                        final RichValueJoiner<? super K, ? super V, ? super VO, ? extendsfinal VR>JoinWindows joinerwindows,
                                  final JoinWindows windowsJoined<K, V, VO> joined);
<VO 

<VT, VR> KStream<K, VR> outerJoinjoin(final KStream<KKTable<K, VO>VT> otherStreamtable,
                             final     final RichValueJoiner<? RichValueJoiner<? super K, ? super V, ? super VOVT, ? extends VR> joiner);
<VT,
 VR> KStream<K, VR> join(final KTable<K, VT> table,
                             final JoinWindows windows,
      RichValueJoiner<? super K, ? super V, ? super VT, ? extends VR> joiner,
                             final Joined<K, V, VO>VT> joined);
 

<VT, VR> KStream<K, VR> joinleftJoin(final KTable<K, VT> table,
                                 final RichValueJoiner<? super K, ? super V, ? super VT, ? extends VR> joiner);
<VT, VR> KStream<K, VR> joinleftJoin(final KTable<K, VT> table,
                                 final RichValueJoiner<? super K, ? super V, ? super VT, ? extends VR> joiner,
                                 final Joined<K, V, VT> joined);
 

<VT<GK, GV, VR>RV> KStream<K, VR>RV> leftJoinjoin(final KTable<KGlobalKTable<GK, VT>GV> tableglobalKTable,
                                 final RichValueJoiner<RichKeyValueMapper<? super K, ? super V, ? super VT, ? extends VR> joiner);
<VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,GK> keyValueMapper,
                                 final RichValueJoiner<? super K, ? super V, ? super VTGV, ? extends VR>RV> joiner);

<GK,
 GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalKTable,
                         final Joined<K, V, VT> joined);
<VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,
 final RichKeyValueMapper<? super K, ? super V, ? extends GK> keyValueMapper,
                                     final RichValueJoiner<? super K, ? super V, ? super VTGV, ? extends VR> joiner,
                  RV> valueJoiner);

KTable

Code Block

language	java

KTable<K, V> filter(final RichPredicate<? super K, ? super V> predicate);
KTable<K, V> filter(final RichPredicate<? super K, ? super V> predicate,
               final Serde<K> keySerde,
   final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
KTable<K, V> filterNot(final RichPredicate<? super K, ? super                  final Serde<V> valSerdeV> predicate);
 

<GKKTable<K, GV, RV> KStream<K, RV> joinV> filterNot(final GlobalKTable<GK,RichPredicate<? GV>super globalKTableK,
 ? super V> predicate,
                       final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);

<VR> KTable<K, VR> mapValues(final KeyValueMapper<RichValueMapper<? super KV, ? superextends VVR, ? extends GK> keyValueMapper,
            super K> mapper);
<VR> KTable<K, VR> mapValues(final RichValueMapper<? super V, ? extends VR, ? super K> mapper,
                     final RichValueJoiner<? super K, ? super V, ? superfinal GVMaterialized<K, ?VR, extends RV> joinerKeyValueStore<Bytes, byte[]>> materialized);
<GK,
<KR> GVKStream<KR, RV> KStream<K, RV> joinV> toStream(final GlobalKTable<GK,RichKeyValueMapper<? GV>super globalKTableK,
 ? super V, ? extends KR> mapper);
 

<KR, VR> KGroupedTable<KR, VR> groupBy(final RichKeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector);
<KR, VR> KGroupedTable<KR,           VR> groupBy(final RichKeyValueMapper<? super K, ? super V, ?KeyValue<KR, extendsVR>> GK> keyValueMapperselector,
                                 final ValueJoiner<? super  V, ? superfinal GVSerialized<KR, ? extends RV> joinerVR> serialized);
<GK 

<VO, GV, RV> KStream<KVR> KTable<K, RV>VR> join(final GlobalKTable<GKKTable<K, GV>VO> globalKTableother,
                            final RichValueJoiner<? super   finalV, RichKeyValueMapper<? super KVO, ? superextends VVR, ? extends GK> keyValueMapper,
      super K> joiner);
<VO, VR> KTable<K, VR> join(final KTable<K, VO> other,
                            final RichValueJoiner<? super KV, ? super VVO, ? superextends GVVR, ? extendssuper RV>K> joiner);



<GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalKTable,
                            final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 

<VO, VR> KTable<K, VR> leftJoin(final KeyValueMapper<? super KKTable<K, ?VO> super V, ? extends GK> keyValueMapper,
     other,
                                final RichValueJoiner<? super KV, ? super VVO, ? superextends GVVR, ? extendssuper RV>K> valueJoinerjoiner);
<GK<VO, GV, RV> KStream<KVR> KTable<K, RV>VR> leftJoin(final GlobalKTable<GKKTable<K, GV>VO> globalKTableother,
                                     final RichKeyValueMapper<ValueJoiner<? super K, ? super V, ? extendssuper GK> keyValueMapperVO,
 ? extends VR>  joiner,
                                final ValueJoiner<? super V, ? super GV, ? extends RV> valueJoiner);
<GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalKTableMaterialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 

<VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                 final RichValueJoiner<? super  finalV, RichKeyValueMapper<? super KVO, ? superextends VVR, ? extends GK> keyValueMapper,
    super K> joiner);
<VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                 final RichValueJoiner<? super KV, ? super VVO, ? superextends GVVR, ? extendssuper RV> valueJoiner);

KTable

Code Block

language	java

KTable<K, V> filter(final RichPredicate<? super K, ? super V> predicate);
KTable<K, V> filter(final RichPredicate<? super K, ? super V> predicate,
K> joiner,
                                       final Materialized<K, VVR, KeyValueStore<Bytes, byte[]>> materialized);

KGroupedStream

Code Block

language	java

KTable<K, V> filterNotreduce(final RichPredicate<? super KRichReducer<V, ? super V> predicateK> reducer);

KTable<K, V> filterNotreduce(final RichPredicate<?RichReducer<V, superK> Kreducer,
 ?  super V> predicate,
                       final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
 

<VR> KTable<K, VR> mapValuesaggregate(final RichValueMapper<?RichInitializer<VR, superK> K, ? super V, ? extends VR> mapper);
<VR> KTable<K, VR> mapValues(final RichValueMapper<initializer,
                             final RichAggregator<? super K, ? super V, ? extends VR> mapperaggregator,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);

<KR><VR> KStream<KRKTable<K, V>VR> toStreamaggregate(final RichKeyValueMapper<?RichInitializer<VR, superK> Kinitializer,
 ? super V, ? extends      KR> mapper);
 

<KR, VR> KGroupedTable<KR, VR> groupBy(final RichKeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector);
<KR, VR> KGroupedTable<KR, VR> groupBy(final RichKeyValueMapper<RichAggregator<? super K, ? super V, VR> aggregator);

SessionWindowedKStream

There are 3 rich interfaces in aggregate() methods. So converting all possible combinations to their rich counterparts can cause a lot of overloads. So, I propose to overload one method with all rich interfaces.

Code Block

language	java

<T> KTable<Windowed<K>, T> aggregate(final RichInitializer<T, Windowed<K>> initializer,
KeyValue<KR, VR>> selector,
                                       final Serialized<KR, VR> serialized);
 

<VO, VR> KTable<K, VR> join(final KTable<K, VO> other,
        final RichAggregator<? super K, ? super V, T> aggregator,
                final RichValueJoiner<? super K, ? super V,                      final RichMerger<? super VOK, ? extends VR> joinerT> sessionMerger);
<VO, VR> KTable<K<VR> KTable<Windowed<K>, VR> joinaggregate(final KTable<KRichInitializer<VR, VO>Windowed<K>> otherinitializer,
                                       final RichValueJoiner<RichAggregator<? super K, ? super V, ?VR> super VOaggregator,
 ? extends VR> joiner,
                            final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 

<VO, VR> KTable<K,final VR> leftJoin(final KTable<KRichMerger<? super K, VO>VR> othersessionMerger,
                                final RichValueJoiner<? super K, ? super V, ? superfinal VOMaterialized<K, ?VR, extends VR> joinerSessionStore<Bytes, byte[]>> materialized);
<VO
KTable<Windowed<K>, VR> KTable<K, VR> leftJoinV> reduce(final KTable<KRichReducer<V, VO> other,
  K> reducer);
KTable<Windowed<K>, V> reduce(final RichReducer<V, K> reducer,
                              final Materialized<K, V, SessionStore<Bytes, byte[]>> materializedAs);


,

TimeWindowedKStream

Code Block

language	java

 
<VR> KTable<Windowed<K>, VR> aggregate(final RichInitializer<VR, K> initializer,
       ValueJoiner<? super K, ? super V, ? super VO, ? extends VR> joiner,
                                final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
 

<VO, VR> KTable<KRichAggregator<? super K, ? super V, VR> aggregator);
<VR> KTable<Windowed<K>, VR> outerJoinaggregate(final KTable<KRichInitializer<VR, VO>K> otherinitializer,
                                 final RichValueJoiner<? super K, ? super V,final RichAggregator<? super VOK, ? extends VR> joiner);
<VOsuper V, VR> KTable<Kaggregator,
  VR> outerJoin(final KTable<K, VO> other,
                                 final RichValueJoiner<? super K, ? super V, ? super VO, ? extends VR> joiner,
   Materialized<K, VR, WindowStore<Bytes, byte[]>> materialized);

KTable<Windowed<K>, V> reduce(final RichReducer<V, K> reducer);
KTable<Windowed<K>, V> reduce(final RichReducer<V, K> reducer,
                              final Materialized<K, VRV, KeyValueStore<BytesWindowStore<Bytes, byte[]>> materialized);

KGroupedStream

KGroupedTable

Code Block

language	java

KTable<K, V> reduce(final RichReducer<K, V> reducer);

 
KTable<K, V> reduce(final RichReducer<KRichReducer<V, V>K> reduceradder,
                    final Materialized<KRichReducer<V, V, K> subtractor,
                    final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);

KTable<K, V> reduce(final RichReducer<V, K> adder,
                    final RichReducer<V, K> subtractor);

<VR> KTable<K, VR> aggregate(final Initializer<VR>RichInitializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> aggregatoradder,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
<VR> KTable<K, VR> aggregate(final RichInitializer<K, VR> initializer,
                             final Aggregator<RichAggregator<? super K, ? super K, ? super V, VR> aggregatorsubtractor,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
<VR> KTable<K, VR> aggregate(final RichInitializer<K, VR>RichInitializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> aggregatoradder,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);



<VR> KTable<K, VR> aggregate(final RichInitializer<K, VR> initializer,
                             final Aggregator<? super K, ? super RichAggregator<? super K, ? super V, VR> aggregator);

<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> aggregator);
<VR> KTable<K, VR> aggregate(final RichInitializer<K, VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> aggregator);

SessionWindowedKStream

There are 3 rich interfaces in aggregate() methods. So converting all possible combinations to their rich counterparts can cause a lot of overloads. So, I propose to overload one method with all rich interfaces.

Code Block

language	java

<T> KTable<Windowed<K>, T> aggregate(final RichInitializer<T> initializer,
                                     final RichAggregator<? super K, ? super V, T> aggregator,
                                     final RichMerger<? super K, T> sessionMerger);
<VR> KTable<Windowed<K>, VR> aggregate(final RichInitializer<VR> initializer,
                                       final RichAggregator<? super K, ? super V, VR> aggregator,
                                       final RichMerger<? super K, VR> sessionMerger,
                                       final Materialized<K, VR, SessionStore<Bytes, byte[]>> materialized);



KTable<Windowed<K>, V> reduce(final RichReducer<K, V> reducer);
KTable<Windowed<K>, V> reduce(final RichReducer<K, V> reducer,
                              final Materialized<K, V, SessionStore<Bytes, byte[]>> materializedAs);

TimeWindowedKStream

Code Block

language	java

 
<VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                       final RichAggregator<? super K, ? super V, VR> aggregator);
<VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer,
                                       final RichAggregator<? super K, ? super V, VR> aggregator,
                                       final Materialized<K, VR, WindowStore<Bytes, byte[]>> materialized);

KTable<Windowed<K>, V> reduce(final RichReducer<K, V> reducer);
KTable<Windowed<K>, V> reduce(final RichReducer<K, V> reducer,
                              final Materialized<K, V, WindowStore<Bytes, byte[]>> materialized);

KGroupedTable

Code Block

language	java

 
KTable<K, V> reduce(final Reducer<V> adder,
                    final RichReducer<K, V> subtractor,
                    final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
KTable<K, V> reduce(final RichReducer<K, V> adder,
                    final Reducer<V> subtractor,
                    final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
KTable<K, V> reduce(final RichReducer<K, V> adder,
                    final RichReducer<K, V> subtractor,
                    final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);



KTable<K, V> reduce(final Reducer<V> adder,
                    final RichReducer<K, V> subtractor);
KTable<K, V> reduce(final RichReducer<K, V> adder,
                    final Reducer<V> subtractor);
KTable<K, V> reduce(final RichReducer<K, V> adder,
                    final RichReducer<K, V> subtractor);



<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final Aggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final Aggregator<? super K, ? super V, VR> subtractor,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor,
                             final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);




<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final Aggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final Aggregator<? super K, ? super V, VR> subtractor);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor);



<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final Aggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor,
                             final Serde<VR> aggValueSerde);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final Aggregator<? super K, ? super V, VR> subtractor,
                             final Serde<VR> aggValueSerde);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor,
                             final Serde<VR> aggValueSerde);



<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final Aggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor,
                             final StateStoreSupplier<KeyValueStore> storeSupplier);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final Aggregator<? super K, ? super V, VR> subtractor,
                             final StateStoreSupplier<KeyValueStore> storeSupplier);
<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
                             final RichAggregator<? super K, ? super V, VR> adder,
                             final RichAggregator<? super K, ? super V, VR> subtractor,
                             final StateStoreSupplier<KeyValueStore> storeSupplier);

Proposed changes

Make record context open to public

Currently we set record context through InternalProcessorContext (StreamTask.updateProcessorContext()) :

Code Block

language	java

 private void updateProcessorContext(final StampedRecord record, final ProcessorNode currNode) {
    processorContext.setRecordContext(new ProcessorRecordContext(record.timestamp, record.offset(), record.partition(), record.topic()));
    processorContext.setCurrentNode(currNode);
}

Thus, the record context is not available in ProcessorContext. As a result, we make the following changes to make it "public"

Code Block

language	java

 
public interface ProcessorContext {
  ...
 
  RecordContext recordContext();
}
 
 
public class ProcessorContextImpl extends AbstractProcessorContext implements RecordCollector.Supplier {
  ...
  @Override
  public RecordContext recordContext() {
    return this.recordContext();
  }
}

Add commit() to RecordContext

We already have RecordContext class with desired set of methods. However, it does not have commit() method. In this KIP we add commit() method to RecordContext interface.

...

language	java

subtractor);

Proposed changes

Move RecordContext from .processor.internals to .processor

Make record context open to public

Currently we set record context through InternalProcessorContext (StreamTask.updateProcessorContext()) :

Code Block

language	java

// the below code snippet already exists, this is just for background. 
private void updateProcessorContext(final StampedRecord record, final ProcessorNode currNode) {
    processorContext.setRecordContext(new ProcessorRecordContext(record.timestamp, record.offset(), record.partition(), record.topic()));
    processorContext.setCurrentNode(currNode);
}

...

Sample processor should look like this:

...

Code Block

language	java

class KStreamKTableJoinProcessor<K1, K2, V1, V2, R> extends AbstractProcessor<K1, V1> {

    ...
    private RecordContext recordContext;       // this line is added in this KIP

    ...

    @Override
	public void process(final K1 key, final V1 value) {

    	recordContext = new RecordContext() {
    		@Override
    		public void commit() {
    // recordContext initialization  		context().commit();
    		}is added in this KIP

    		@Override
    		public long offset() {
        		return context().recordContext().offset();
    		}

    		@Override
    		public long timestamp() {
        		return context().recordContext().timestamp();
    		}

    		@Override
    		public String topic() {
        		return context().recordContext().topic();
    		}

    		@Override
    		public int partition() {
        		return context().recordContext().partition();
    		}
      };

	
 
    if (key != null && value != null) {
        final V2 value2 = valueGetter.get(keyMapper.apply(key, value));
        if (leftJoin || value2 != null) {
            context().forward(key, joiner.apply(value, value2, recordContext));    
        }
    }
}


}

...

Space shortcuts

Child pages

Versions Compared

Old Version 29

New Version Current

Key

Rich Interfaces

Public Interfaces

KTable

KTable

KGroupedStream

SessionWindowedKStream

TimeWindowedKStream

KGroupedStream

KGroupedTable

SessionWindowedKStream

TimeWindowedKStream

KGroupedTable

Proposed changes

Make record context open to public

Add commit() to RecordContext

Proposed changes

Move `RecordContext` from `.processor.internals` to `.processor`

Make record context open to public

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 29

New Version Current

Key

Rich Interfaces

Public Interfaces

KTable

KTable

KGroupedStream

SessionWindowedKStream

TimeWindowedKStream

KGroupedStream

KGroupedTable

SessionWindowedKStream

TimeWindowedKStream

KGroupedTable

Proposed changes

Make record context open to public

Add commit() to RecordContext

Proposed changes

Move RecordContext from .processor.internals to .processor

Make record context open to public

Move `RecordContext` from `.processor.internals` to `.processor`