Skip to content

Commit 8059503

Browse files
aliehsaeediialiehsaeedii
authored
KAFKA-20132: Implement TimestampedKeyValueStoreWithHeaders (3/N) (apache#21451)
This PR implements the metered layer of the TimestampedKeyValueStoreWithHeaders introduced in KIP-1271. Reviewers: TengYao Chi <frankvicky@apache.org>, Matthias J. Sax <matthias@confluent.io>
1 parent 4297d99 commit 8059503

4 files changed

Lines changed: 648 additions & 4 deletions

File tree

‎streams/src/main/java/org/apache/kafka/streams/state/TimestampedKeyValueStoreWithHeaders.java‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.kafka.streams.state;
18+
19+
/**
20+
* A key-(value/timestamp/headers) store that supports put/get/delete.
21+
*
22+
* @param <K> The key type
23+
* @param <V> The value type
24+
*/
25+
public interface TimestampedKeyValueStoreWithHeaders<K, V>
26+
extends KeyValueStore<K, ValueTimestampHeaders<V>> {
27+
}

‎streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package org.apache.kafka.streams.state.internals;
1818

1919
import org.apache.kafka.common.TopicPartition;
20+
import org.apache.kafka.common.header.internals.RecordHeaders;
2021
import org.apache.kafka.common.metrics.Sensor;
2122
import org.apache.kafka.common.serialization.Serde;
2223
import org.apache.kafka.common.serialization.Serializer;
@@ -86,7 +87,7 @@ public class MeteredKeyValueStore<K, V>
8687
private final String metricsScope;
8788
protected final Time time;
8889
protected Sensor putSensor;
89-
private Sensor putIfAbsentSensor;
90+
protected Sensor putIfAbsentSensor;
9091
protected Sensor getSensor;
9192
protected Sensor deleteSensor;
9293
private Sensor putAllSensor;
@@ -323,7 +324,7 @@ public void put(final K key,
323324
final V value) {
324325
Objects.requireNonNull(key, "key cannot be null");
325326
try {
326-
maybeMeasureLatency(() -> wrapped().put(keyBytes(key), serdes.rawValue(value)), time, putSensor);
327+
maybeMeasureLatency(() -> wrapped().put(keyBytes(key), serdes.rawValue(value, new RecordHeaders())), time, putSensor);
327328
maybeRecordE2ELatency();
328329
} catch (final ProcessorStateException e) {
329330
final String message = String.format(e.getMessage(), key, value);
@@ -420,11 +421,11 @@ public void close() {
420421
}
421422

422423
protected V outerValue(final byte[] value) {
423-
return value != null ? serdes.valueFrom(value) : null;
424+
return value != null ? serdes.valueFrom(value, new RecordHeaders()) : null;
424425
}
425426

426427
protected Bytes keyBytes(final K key) {
427-
return Bytes.wrap(serdes.rawKey(key));
428+
return Bytes.wrap(serdes.rawKey(key, new RecordHeaders()));
428429
}
429430

430431
private List<KeyValue<Bytes, byte[]>> innerEntries(final List<KeyValue<K, V>> from) {

‎streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.kafka.streams.state.internals;
18+
19+
import org.apache.kafka.common.header.Headers;
20+
import org.apache.kafka.common.header.internals.RecordHeaders;
21+
import org.apache.kafka.common.serialization.Serde;
22+
import org.apache.kafka.common.utils.Bytes;
23+
import org.apache.kafka.common.utils.Time;
24+
import org.apache.kafka.streams.errors.ProcessorStateException;
25+
import org.apache.kafka.streams.processor.internals.SerdeGetter;
26+
import org.apache.kafka.streams.query.Position;
27+
import org.apache.kafka.streams.query.PositionBound;
28+
import org.apache.kafka.streams.query.Query;
29+
import org.apache.kafka.streams.query.QueryConfig;
30+
import org.apache.kafka.streams.query.QueryResult;
31+
import org.apache.kafka.streams.state.KeyValueStore;
32+
import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
33+
import org.apache.kafka.streams.state.ValueTimestampHeaders;
34+
35+
import java.util.Objects;
36+
37+
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency;
38+
39+
40+
/**
41+
* A Metered {@link TimestampedKeyValueStoreWithHeaders} wrapper that is used for recording operation metrics, and hence
42+
* its inner KeyValueStore implementation does not need to provide its own metrics collecting functionality.
43+
*
44+
* The inner {@link KeyValueStore} of this class is of type &lt;Bytes, byte[]&gt;,
45+
* hence we use {@link Serde}s to convert from &lt;K, ValueTimestampHeaders&lt;V&gt;&gt; to &lt;Bytes, byte[]&gt;.
46+
*
47+
* @param <K> key type
48+
* @param <V> value type (wrapped in {@link ValueTimestampHeaders})
49+
*/
50+
public class MeteredTimestampedKeyValueStoreWithHeaders<K, V>
51+
extends MeteredKeyValueStore<K, ValueTimestampHeaders<V>>
52+
implements TimestampedKeyValueStoreWithHeaders<K, V> {
53+
54+
MeteredTimestampedKeyValueStoreWithHeaders(final KeyValueStore<Bytes, byte[]> inner,
55+
final String metricScope,
56+
final Time time,
57+
final Serde<K> keySerde,
58+
final Serde<ValueTimestampHeaders<V>> valueSerde) {
59+
super(inner, metricScope, time, keySerde, valueSerde);
60+
}
61+
62+
@SuppressWarnings("unchecked")
63+
@Override
64+
protected Serde<ValueTimestampHeaders<V>> prepareValueSerdeForStore(final Serde<ValueTimestampHeaders<V>> valueSerde,
65+
final SerdeGetter getter) {
66+
if (valueSerde == null) {
67+
return new ValueTimestampHeadersSerde<>((Serde<V>) getter.valueSerde());
68+
} else {
69+
return super.prepareValueSerdeForStore(valueSerde, getter);
70+
}
71+
}
72+
73+
@Override
74+
public void put(final K key,
75+
final ValueTimestampHeaders<V> value) {
76+
Objects.requireNonNull(key, "key cannot be null");
77+
try {
78+
final Headers headers = value != null ? value.headers() : new RecordHeaders();
79+
maybeMeasureLatency(() -> wrapped().put(keyBytes(key, headers), serdes.rawValue(value, headers)), time, putSensor);
80+
maybeRecordE2ELatency();
81+
} catch (final ProcessorStateException e) {
82+
final String message = String.format(e.getMessage(), key, value);
83+
throw new ProcessorStateException(message, e);
84+
}
85+
}
86+
87+
@Override
88+
public ValueTimestampHeaders<V> putIfAbsent(final K key,
89+
final ValueTimestampHeaders<V> value) {
90+
Objects.requireNonNull(key, "key cannot be null");
91+
final Headers headers = value != null ? value.headers() : new RecordHeaders();
92+
final ValueTimestampHeaders<V> currentValue = maybeMeasureLatency(
93+
() -> outerValue(wrapped().putIfAbsent(keyBytes(key, headers), serdes.rawValue(value, headers))),
94+
time,
95+
putIfAbsentSensor
96+
);
97+
maybeRecordE2ELatency();
98+
return currentValue;
99+
}
100+
101+
@Override
102+
public <R> QueryResult<R> query(final Query<R> query,
103+
final PositionBound positionBound,
104+
final QueryConfig config) {
105+
throw new UnsupportedOperationException("Querying is not supported for " + getClass().getSimpleName());
106+
}
107+
108+
@Override
109+
public Position getPosition() {
110+
throw new UnsupportedOperationException("Position is not supported for " + getClass().getSimpleName());
111+
}
112+
113+
protected Bytes keyBytes(final K key, final Headers headers) {
114+
return Bytes.wrap(serdes.rawKey(key, headers));
115+
}
116+
117+
}

0 commit comments

Comments
 (0)