...
JIRA:
Jira | ||||||
---|---|---|---|---|---|---|
|
Jira | ||||||
---|---|---|---|---|---|---|
|
Jira | ||||||||
---|---|---|---|---|---|---|---|---|
|
Jira | ||||||||
---|---|---|---|---|---|---|---|---|
|
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
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:
- on-demand
commit()
feature - having access to RecordContext within an operator
- having access to a read-only key for ValueJoiner, ValueTransformer, ValueMapper interfaces
...
Code Block | ||
---|---|---|
| ||
public interface RichInitializer<V, K> {
V apply(K key);
}
public interface RichValueMapper<V, 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<V, K> {
V apply(final V value1, final V value2, final K key, 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<V, VR, K> {
void init(final ProcessorContext context);
VR transform(final V value, final K key);
@Deprecated
VR punctuate(final long timestamp);
void close();
}
public interface RichValueTransformerSupplier<V, VR, K> {
RichValueTransformer<V, VR, K> get();
}
|
...
Code Block | ||
---|---|---|
| ||
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 V, ? extends VR, ? super 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 V, ? extends Iterable<? extends VR>, ? super 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 V, ? extends VR, ? super 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 V, ? super VO, ? extends VR, ? super K> joiner, final JoinWindows windows, final Joined<K, V, VO> joined); <VO, VR> KStream<K, VR> joinleftJoin(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> joiner, final JoinWindows windows, final Joined<K, V, VO> joinedJoinWindows windows); <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> joiner, final JoinWindows windows); <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream, final RichValueJoiner<? superJoined<K, V, ? super VO, ? extends VR, ? super K> joiner, VO> joined); <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream, final JoinWindows windows RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> joiner, final Joined<K, V, VO> joinedJoinWindows windows); <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> joiner, final JoinWindows windows); <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream, final RichValueJoiner<?Joined<K, super V, ? super VO, ? extends VR, ? super K> joinerVO> joined); <VT, VR> KStream<K, VR> join(final KTable<K, VT> table, final RichValueJoiner<? super K, final JoinWindows windows, ? super V, ? super VT, ? extends VR> joiner); <VT, VR> KStream<K, VR> join(final KTable<K, VT> table, final Joined<K,RichValueJoiner<? super K, ? super V, VO> joined)? super VT, ? extends VR> joiner, final Joined<K, V, 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, ? superextends VT, ? extends VR> joiner); <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> tableGK> 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 | ||
---|---|---|
| ||
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 V, ? extends VR, ? super K, K> mapper); <VR> KTable<K, VR> mapValues(final RichValueMapper<? super V, ? extends GK> keyValueMapperVR, ? super K> mapper, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized); <KR> KStream<KR, V> toStream(final RichValueJoiner<RichKeyValueMapper<? super K, ? super V, ? superextends GV, ? extends RV> joiner); <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalKTable, 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, ? super GVfinal Serialized<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, ? extendssuper GK> keyValueMapper, 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, final KeyValueMapper<? super K, ? super V, ? extends GK> keyValueMapper, VR> leftJoin(final KTable<K, VO> 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 | ||
---|---|---|
| ||
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 | ||
---|---|---|
| ||
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> V, ? extends VR, ? super K> mapper); <VR> KTable<K, VR> mapValues(final RichValueMapper<initializer, final RichAggregator<? super VK, ? extendssuper VRV, ? super K> mapperVR> aggregator, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized); <KR><VR> KStream<KRKTable<K, V>VR> toStreamaggregate(final RichKeyValueMapper<?RichInitializer<VR, superK> K, ? 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<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 | ||
---|---|---|
| ||
<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 V, ? super VO, ? extends VR, final RichMerger<? super K> joinerK, T> sessionMerger); <VO, VR> KTable<K<VR> KTable<Windowed<K>, VR> joinaggregate(final KTable<KRichInitializer<VR, VO>Windowed<K>> otherinitializer, final RichValueJoiner<RichAggregator<? super VK, ? super VOV, ?VR> extends VRaggregator, ? super K> 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 V, ? super VO, ?final extendsMaterialized<K, VR, ? super K> 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 | ||
---|---|---|
| ||
<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 V, ? super VO,final RichAggregator<? extendssuper VRK, ? super K> joiner); <VOV, VR> KTable<Kaggregator, VR> outerJoin(final KTable<K, VO> other, final RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> 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); |
KGroupedTable
KGroupedStream
Code Block | ||
---|---|---|
| ||
KTable<K, V> reduce(final RichReducer<V, K> reducer); KTable<K, V> reduce(adder, final RichReducer<V, K> reducersubtractor, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized); <VR> KTable<K, VR>V> aggregatereduce(final Initializer<VR> initializerRichReducer<V, K> adder, final RichAggregator<? super KRichReducer<V, ? super VK> subtractor); <VR> KTable<K, VR> aggregator aggregate(final RichInitializer<VR> initializer, final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized); <VR> KTable<KRichAggregator<? super K, ? super V, VR> aggregate(final RichInitializer<K, VR> initializer,adder, final Aggregator<RichAggregator<? 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 | ||
---|---|---|
| ||
<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<V, K> reducer);
KTable<Windowed<K>, V> reduce(final RichReducer<V, K> reducer,
final Materialized<K, V, SessionStore<Bytes, byte[]>> materializedAs);
|
TimeWindowedKStream
Code Block | ||
---|---|---|
| ||
<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<V, K> reducer);
KTable<Windowed<K>, V> reduce(final RichReducer<V, K> reducer,
final Materialized<K, V, WindowStore<Bytes, byte[]>> materialized);
|
KGroupedTable
Code Block | ||
---|---|---|
| ||
KTable<K, V> reduce(final Reducer<V> adder,
final RichReducer<V, K> subtractor,
final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
KTable<K, V> reduce(final RichReducer<V, K> adder,
final Reducer<V> subtractor,
final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
KTable<K, V> reduce(final RichReducer<V, K> adder,
final RichReducer<V, K> subtractor,
final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized);
KTable<K, V> reduce(final Reducer<V> adder,
final RichReducer<V, K> subtractor);
KTable<K, V> reduce(final RichReducer<V, K> adder,
final Reducer<V> subtractor);
KTable<K, V> reduce(final RichReducer<V, K> adder,
final RichReducer<V, K> 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
StreamTask.updateProcessorContext()
) :Code Block | ||
---|---|---|
| ||
// 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);
} |
Thus, the record context is not available in ProcessorContext interface. As a result, we make the following changes to make it "public"
Code Block | ||
---|---|---|
| ||
public interface ProcessorContext {
...
...
RecordContext recordContext(); // this line is added in this KIP
}
public class ProcessorContextImpl extends AbstractProcessorContext implements RecordCollector.Supplier {
...
@Override
public RecordContext recordContext() { // only this method is added in this KIP
return this.recordContext();
}
}
|
Add commit() to RecordContext
Currently RecordContext
interface have most of the desired set of methods required in this KIP.
However, it does not have commit()
method. In this KIP we add commit()
method to RecordContext
interface.
Because ProcessorRecordContext
implements RecordContext
, we inherit newly added commit()
method in ProcessorRecordContext
interface.
However, call to a commit()
method, is valid only within RecordContext
interface (at least for now), we throw an exception in ProcessorRecordContext.commit()
.
...
language | java |
---|
subtractor);
|
Proposed changes
Move
RecordContext
from.
processor.internals
to.processor
Make record context open to public
StreamTask.updateProcessorContext()
) :Code Block | ||
---|---|---|
| ||
// 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 | ||
---|---|---|
| ||
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() { // recordContext initialization is added in this KIP
@Override
public void commit() {
context().commit();
}
@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));
}
}
}
} |
...