KAFKA-20158: Adding MeteredSessionStore and tests (3/N)#21513
KAFKA-20158: Adding MeteredSessionStore and tests (3/N)#21513bbejeck wants to merge 11 commits intoapache:trunkfrom
Conversation
bbejeck
changed the title
aliehsaeedii
left a comment
aliehsaeedii
left a comment
There was a problem hiding this comment.
Thanks @bbejeck. I made a pass. If inheritance and therefore refactoring is possible, then reviewing would be easier.
...s/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java
Outdated
Show resolved
Hide resolved
...s/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java
Outdated
Show resolved
Hide resolved
bbejeck
force-pushed
the
KAFKA-20158_session_store_w_headers_part_3
branch
from
f948830 to
85a1bb6
Compare
| final Time time) { | ||
| super(inner, metricsScope, keySerde, createAggregationWithHeadersSerde(aggSerde), time); | ||
| this.rawAggSerde = aggSerde; | ||
| } |
There was a problem hiding this comment.
Wondering if it could be
MeteredSessionStoreWithHeaders(final SessionStore<Bytes, byte[]> inner,
final String metricsScope,
final Serde<K> keySerde,
final Serde<AggregationWithHeaders<AGG>> aggSerde,
final Time time) {
super(inner, metricsScope, keySerde, aggSerde, time);
this.rawAggSerde = aggSerde;
}
There was a problem hiding this comment.
If it's doable, then we dont need the properties and most of the methods except the write ones. Of course we must change MeteredSessionStore to send empty headers to keyBytes for read methods as well.
| final Time time) { | ||
| super(inner, metricsScope, keySerde, createAggregationWithHeadersSerde(aggSerde), time); | ||
| this.rawAggSerde = aggSerde; | ||
| } |
There was a problem hiding this comment.
If it's doable, then we dont need the properties and most of the methods except the write ones. Of course we must change MeteredSessionStore to send empty headers to keyBytes for read methods as well.
| protected Sensor restoreSensor; | ||
|
|
||
| protected final LongAdder numOpenIterators = new LongAdder(); | ||
| protected final NavigableSet<MeteredIterator> openIterators = new ConcurrentSkipListSet<>(Comparator.comparingLong(MeteredIterator::startTimestamp)); |
There was a problem hiding this comment.
Let's send empty headers to keyBytes or any de/serializer caller of this class as we are depcrating the old ones with no headers an input. Thanks
...s/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java
Outdated
Show resolved
Hide resolved
|
|
||
| @SuppressWarnings("unchecked") | ||
| @Override | ||
| public boolean setFlushListener(final CacheFlushListener<Windowed<K>, AggregationWithHeaders<AGG>> listener, |
There was a problem hiding this comment.
| public boolean setFlushListener(final CacheFlushListener<Windowed<K>, AggregationWithHeaders<AGG>> listener, | |
| public boolean setFlushListener(final CacheFlushListener<Windowed<K>, final AggregationWithHeaders<AGG>> listener, |
There was a problem hiding this comment.
@frankvicky the CacheFlushListener is decalred final the other part there is within the generics definition.
bbejeck
force-pushed
the
KAFKA-20158_session_store_w_headers_part_3
branch
from
71c27e9 to
c266f45
Compare
|
@aliehsaeedii @frankvicky all comments addressed - I took another pass and was able to clean things up quite a bit |
3 participants
This PR adds
MeteredSessionStoreWithHeadersand a unit test forsupporting KIP-1271. This is a stacked PR only review
c266f45