...
JIRA:
Jira | ||||||
---|---|---|---|---|---|---|
|
Jira | ||||||||
---|---|---|---|---|---|---|---|---|
| ||||||||
Jira | ||||||||
|
Jira | ||||||||
---|---|---|---|---|---|---|---|---|
|
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 an atomic a hybrid solution to rich functions in Streams and accessing read-only keys within ValueJoiner, 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
XXXWithKey
interfaces
Public Interfaces
The following methods will be added to related classes.
KStream.java
- ValueJoiner, ValueTransformer, ValueMapper interfaces
Rich 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);
void close();
}
public interface RichValueTransformerSupplier<V, VR, K> {
RichValueTransformer<V, VR, K> get();
}
|
Public Interfaces
KStream
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, 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> 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>> processor); 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); <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 Serde<KR> keySerde, final Serde<V> valSerde); <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR> joiner, final JoinWindows windows); <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR> joiner, final JoinWindows windows, final Serde<K> keySerde, 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 Serde<V> thisValueSerdeSerialized<KR, final Serde<VO> otherValueSerdeV> serialized); <VO, VR> KStream<K, VR> leftJoinjoin(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR> joinerVR, ? super K> joiner, final JoinWindows windows); <VO, VR> KStream<K, VR> leftJoinjoin(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR> joinerVR, ? super K> joiner, final JoinWindows windows, final Joined<K, V, final Serde<K> keySerde,VO> joined); <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream, final Serde<V> thisValSerde RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> joiner, final Serde<VO>JoinWindows otherValueSerdewindows); <VO, VR> KStream<K, VR> outerJoinleftJoin(final KStream<K, VO> otherStream, final RichValueJoiner<? super V, ? super VO, ? extends VR> VR, ? super 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 RichValueJoiner<? super VKStream<K, ? super VO, ? extends VR> joinerVO> otherStream, final RichValueJoiner<? JoinWindowssuper windowsV, ? super VO, ? extends VR, ? super K> joiner, final Serde<K> keySerde, final JoinWindows windows); <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream, final Serde<V> thisValueSerde, final RichValueJoiner<? super V, ? super VO, ? extends VR, ? super K> joiner, final Serde<VO> otherValueSerde); <VT, VR> KStream<K, VR> join(final KTable<K, VT> table final JoinWindows windows, final RichValueJoiner<? super V, ? superfinal VTJoined<K, ?V, extendsVO> VR> joinerjoined); <VT, VR> KStream<K, VR> join(final KTable<K, VT> table, final RichValueJoiner<? super VK, ? super VTV, ? super VT, ? extends VR> joiner); <VT, VR> KStream<K, VR> join(final KTable<K, VT> table, final Serde<K> keySerde RichValueJoiner<? super K, ? super V, ? super VT, ? extends VR> joiner, final Serde<V> valSerde Joined<K, V, VT> joined); <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table, final RichValueJoiner<? super K, ? super V, ? super VT, ? extends VR> joiner); <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table, final RichValueJoiner<? super K, ? super V, ? super VT, ? extends VR> joiner, final Serde<K>Joined<K, keySerdeV, final Serde<V> valSerdeVT> joined); <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalKTable, final RichKeyValueMapper<? super K, ? super V, ? extends GK> keyValueMapper, final RichValueJoiner<? super K, ? super V, ? super GV, ? extends RV> joiner); <GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalKTable, final RichKeyValueMapper<? super K, ? super V, ? extends GK> keyValueMapper, final RichValueJoiner<? super K, ? super V, ? super GV, ? extends 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 String queryableStoreName); KTable<K, V> filter(final RichPredicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized); KTable<K, V> filterNot(final RichPredicate<? super K, ? super V> predicate); KTable<K, V> filterNot(final RichPredicate<? super K, ? super V> predicate, final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> filterNot(final RichPredicate<? super K, ? super V> predicate, final String queryableStoreName); final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized); <VR> KTable<K, VR> mapValues(final RichValueMapper<? super V, ? extends VR> VR, ? super K> mapper); <VR> KTable<K, VR> mapValues(final RichValueMapper<? super V, ? extends VR> mapper, final Serde<VR> valueSerde, final String queryableStoreName); <VR> KTable<K, VR> mapValues(final RichValueMapper<VR, ? super V, ? extends VR> K> mapper, final Serde<VR>Materialized<K, valueSerdeVR, KeyValueStore<Bytes, byte[]>> materialized); <KR> KStream<KR, V> toStream(final RichKeyValueMapper<? super K, ? super V, ? extends KR> mapper); <KR, VR> KGroupedTable<KR, final StateStoreSupplier<KeyValueStore> storeSupplier); void foreach(final RichForeachAction<? super K, ? super V> action); <KR> KStream<KR, V> toStreamVR> groupBy(final RichKeyValueMapper<? super K, ? super V, ?KeyValue<KR, extendsVR>> KR> mapperselector); <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, VR>> selector, , final Serde<KR> keySerdeSerialized<KR, final Serde<VR> valueSerdeVR> serialized); <VO, VR> KTable<K, VR> join(final KTable<K, VO> other, final RichValueJoiner<? super V, ? super VO, ? extends VR> VR, ? super K> joiner); <VO, VR> KTable<K, VR> join(final KTable<K, VO> other, final RichValueJoiner<? super V, ? super VO, ? extends VR> VR, ? super K> joiner, final Serde<VR>Materialized<K, joinSerdeVR, KeyValueStore<Bytes, final String queryableStoreNamebyte[]>> materialized); <VO, VR> KTable<K, VR> joinleftJoin(final KTable<K, VO> other, final RichValueJoiner<? super V, ? super VO, ? extends VR> joinerVR, ? super final StateStoreSupplier<KeyValueStore> storeSupplierK> joiner); <VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other, final RichValueJoiner<ValueJoiner<? super K, ? super V, ? super VO, ? extends VR> joiner); <VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other, final RichValueJoiner<? super V, ? super VO, ? extends VR> joiner, Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized); <VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other, final Serde<VR> joinSerde, final RichValueJoiner<? super V, ? super VO, ? extends VR, ? super final String queryableStoreNameK> joiner); <VO, VR> KTable<K, VR> leftJoinouterJoin(final KTable<K, VO> other, final RichValueJoiner<? super V, ? super VO, ? extends VR> VR, ? super K> joiner, final StateStoreSupplier<KeyValueStore> storeSupplier); <VOMaterialized<K, VR> KTable<KVR, VR> outerJoin(final KTable<KKeyValueStore<Bytes, VO> other, byte[]>> materialized); |
KGroupedStream
Code Block | ||
---|---|---|
| ||
KTable<K, V> reduce(final RichReducer<V, K> reducer); KTable<K, V> reduce(final RichReducer<V, K> reducer, final RichValueJoiner<? super V, ? superfinal VOMaterialized<K, ?V, extends VR> joinerKeyValueStore<Bytes, byte[]>> materialized); <VO,<VR> VR> KTable<K, VR> outerJoinaggregate(final KTable<KRichInitializer<VR, VO>K> otherinitializer, final RichAggregator<? super finalK, RichValueJoiner<? super V, ? super VO, ? extends VR> joineraggregator, final Serde<VR>Materialized<K, joinSerdeVR, KeyValueStore<Bytes, final String queryableStoreNamebyte[]>> materialized); <VO,<VR> VR> KTable<K, VR> outerJoinaggregate(final KTable<KRichInitializer<VR, VO>K> otherinitializer, final RichValueJoiner<RichAggregator<? super VK, ? 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>> initializerVO, ? extends VR> joiner, final StateStoreSupplier<KeyValueStore> storeSupplier); |
KGroupedStream
Code Block | ||
---|---|---|
| ||
KTable<K, V> reduce(final RichReducer<V> reducer); KTable<K, V> reduce(final RichReducer<V> reducer final RichAggregator<? super K, ? super V, T> aggregator, final String queryableStoreName); KTable<K, V> reduce(final RichReducer<V> reducer, final RichMerger<? super K, T> sessionMerger); <VR> KTable<Windowed<K>, VR> aggregate(final StateStoreSupplier<KeyValueStore> storeSupplier); <W extends Window> KTable<Windowed<K>RichInitializer<VR, V> reduce(final RichReducer<V> reducerWindowed<K>> initializer, final RichAggregator<? super K, ? super V, final Windows<W> windows, VR> aggregator, final String queryableStoreName); <W extends Window> KTable<Windowed<K>, V> reduce(final RichReducer<V> reducerRichMerger<? super K, VR> sessionMerger, final Materialized<K, VR, SessionStore<Bytes, byte[]>> materialized); KTable<Windowed<K>, V> reduce(final RichReducer<V, K> final Windows<W> windowsreducer); <W extends Window> KTable<Windowed<K>, V> reduce(final RichReducer<V, 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, final Windows<W> windows, final RichAggregator<? super K, ? super V, final StateStoreSupplier<WindowStore> storeSupplierVR> aggregator); <VR> KTable<Windowed<K>, V>VR> reduceaggregate(final RichReducer<V> reducerRichInitializer<VR, K> initializer, final SessionWindows sessionWindows, final RichAggregator<? super K, ? super V, VR> aggregator, final String queryableStoreName); KTable<Windowed<K>, V> reduce(final RichReducer<V> reducer, final Materialized<K, VR, WindowStore<Bytes, byte[]>> materialized); KTable<Windowed<K>, V> reduce(final RichReducer<V, final SessionWindows sessionWindowsK> reducer); KTable<Windowed<K>, V> reduce(final RichReducer<V, RichReducer<V>K> reducer, final SessionWindowsMaterialized<K, sessionWindowsV, WindowStore<Bytes, byte[]>> materialized); |
KGroupedTable
Code Block | ||
---|---|---|
| ||
KTable<K, V> reduce(final RichReducer<V, K> adder, final StateStoreSupplier<SessionStore> storeSupplier); <VR> KTable<KRichReducer<V, VR> aggregate(final Initializer<VR> initializerK> subtractor, final Materialized<K, V, KeyValueStore<Bytes, byte[]>> materialized); KTable<K, V> reduce(final RichAggregator<? super KRichReducer<V, ?K> super V, VR> aggregator,adder, final RichReducer<V, K> subtractor); <VR> KTable<K, VR> aggregate(final Serde<VR>RichInitializer<VR> aggValueSerdeinitializer, final String queryableStoreName); <VR> KTable<K, VR> aggregate(final Initializer<VR> initializerRichAggregator<? super K, ? super V, VR> adder, final RichAggregator<? super K, ? super V, VR> aggregatorsubtractor, final Serde<VR>Materialized<K, aggValueSerdeVR, KeyValueStore<Bytes, final String queryableStoreName); byte[]>> materialized); <VR> KTable<K, VR> aggregate(final Initializer<VR>RichInitializer<VR> initializer, final Aggregator<RichAggregator<? super K, ? super V, VR> aggregator, final Serde<VR> aggValueSerde); <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer, final RichAggregator<? super K, ? super V, VR> aggregator, final Serde<VR> aggValueSerde); <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer, final RichAggregator<? super K, ? super V, VR> aggregator, final StateStoreSupplier<KeyValueStore> storeSupplier); <VR> KTable<K, VR> aggregate(final Initializer<VR> initializer, final RichAggregator<? super K, ? super V, VR> aggregator, final StateStoreSupplier<KeyValueStore> storeSupplier); <W extends Window, VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer, final RichAggregator<? super K, ? super V, VR> aggregator, final Windows<W> windows, final Serde<VR> aggValueSerde, final String queryableStoreName); <W extends Window, VR> KTable<Windowed<K>, VR> aggregate(final Initializer<VR> initializer, final RichAggregator<? super K, ? super V, VR> aggregator, final Windows<W> windows, final StateStoreSupplier<WindowStore> storeSupplier); <T> KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer, final RichAggregator<? super K, ? super V, T> aggregator, final Merger<? super K, T> sessionMerger, final SessionWindows sessionWindows, final Serde<T> aggValueSerde, final String queryableStoreName); <T> KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer, final RichAggregator<? super K, ? super V, T> aggregator, final Merger<? super K, T> sessionMerger, final SessionWindows sessionWindows, final Serde<T> aggValueSerde); <T> KTable<Windowed<K>, T> aggregate(final Initializer<T> initializer, final RichAggregator<? super K, ? super V, T> aggregator, final Merger<? super K, T> sessionMerger, final SessionWindows sessionWindows, final Serde<T> aggValueSerde, final StateStoreSupplier<SessionStore> storeSupplier); |
KGroupedTable
Code Block | ||
---|---|---|
| ||
KTable<K, V> reduce(final RichReducer<V> adder,
final Reducer<V> subtractor,
final String queryableStoreName);
KTable<K, V> reduce(final Reducer<V> adder,
final RichReducer<V> subtractor,
final String queryableStoreName);
KTable<K, V> reduce(final RichReducer<V> adder,
final RichReducer<V> subtractor,
final String queryableStoreName);
KTable<K, V> reduce(final RichReducer<V> adder,
final Reducer<V> subtractor);
KTable<K, V> reduce(final Reducer<V> adder,
final RichReducer<V> subtractor);
KTable<K, V> reduce(final RichReducer<V> adder,
final RichReducer<V> subtractor);
KTable<K, V> reduce(final RichReducer<V> adder,
final Reducer<V> subtractor,
final StateStoreSupplier<KeyValueStore> storeSupplier);
KTable<K, V> reduce(final Reducer<V> adder,
final RichReducer<V> subtractor,
final StateStoreSupplier<KeyValueStore> storeSupplier);
KTable<K, V> reduce(final RichReducer<V> adder,
final RichReducer<V> 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 String queryableStoreName);
<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 RichAggregator<? super K, ? super V, VR> adder,
final RichAggregator<? super K, ? super V, VR> subtractor,
final Serde<VR> aggValueSerde,
final String queryableStoreName);
<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 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 | ||
---|---|---|
| ||
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. We make the following changes to make it "public"
Code Block | ||
---|---|---|
| ||
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.
Code Block | ||
---|---|---|
| ||
public interface RecordContext {
. . .
void commit ();
}
public class ProcessorRecordContext implements RecordContext {
. . .
@Override
void commit () {
throw new UnsupportedOperationException("commit() is not supported in this context");
}
}
|
Sample processor should look like this:
adder,
final RichAggregator<? super K, ? super V, VR> 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 | ||
Code Block | ||
| ||
class KStreamKTableJoinProcessor<K1, K2, V1, V2, R> extends AbstractProcessor<K1, V1> { ... private RecordContext recordContext; ... @Override public void process(final K1 key, final V1 value) { recordContext = new RecordContext() { @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 intlong partitionoffset() { return context().recordContext().partitionoffset(); } }; if (key != null && value != null) { final V2 value2 = valueGetter.get(keyMapper.apply(key, value)); if (leftJoin || value2 != null @Override public long timestamp() { return context().forward(key, joiner.apply(value, value2, recordContext));recordContext().timestamp(); } } @Override } } } |
Rich Interfaces
Code Block | ||
---|---|---|
| ||
public interface RichValueMapper<V, VR> { VR apply(final V value, final RecordContext recordContext); } public interface RichValueJoiner<V1, V2, VR> { VR apply(final V1 value1, final V2 value2, final RecordContext recordContext); } public interface RichKeyValueMapper<K, V, VR> { VR apply(final K key, final V value, final RecordContext recordContext); } public interface RichReducer<V> { V apply(final V value1, final V value2, 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 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)); } } } } |
Rejected Alternatives
Not yet.
...