Skip to content

[MINOR] In-memory stores cleanup #6595

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 6 commits into from
Apr 26, 2019
Merged

[MINOR] In-memory stores cleanup #6595

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3d54963
Removed unnecessary this and initialized data structures in declaration
ableegoldman Apr 17, 2019
664404c
Added missing types to KeyValueIterators in store tests
ableegoldman Apr 17, 2019
af5e829
Safely register new iterators, iterator classes should be static
ableegoldman Apr 17, 2019
cee2e5d
Don't modify keyFrom/keyTo
ableegoldman Apr 18, 2019
a41ba64
Use static emptyIterator() method instead of dummy constructor
ableegoldman Apr 18, 2019
fcd3200
Fixed checkstyle
ableegoldman Apr 19, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 13 additions & 15 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,20 +33,18 @@

public class InMemoryKeyValueStore implements KeyValueStore<Bytes, byte[]> {
private final String name;
private final ConcurrentNavigableMap<Bytes, byte[]> map;
private final ConcurrentNavigableMap<Bytes, byte[]> map = new ConcurrentSkipListMap<>();
private volatile boolean open = false;

private static final Logger LOG = LoggerFactory.getLogger(InMemoryKeyValueStore.class);

public InMemoryKeyValueStore(final String name) {
this.name = name;

this.map = new ConcurrentSkipListMap<>();
}

@Override
public String name() {
return this.name;
return name;
}

@Override
Expand All @@ -65,7 +63,7 @@ public void init(final ProcessorContext context,
});
}

this.open = true;
open = true;
}

@Override
Expand All @@ -75,20 +73,20 @@ public boolean persistent() {

@Override
public boolean isOpen() {
return this.open;
return open;
}

@Override
public byte[] get(final Bytes key) {
return this.map.get(key);
return map.get(key);
}

@Override
public void put(final Bytes key, final byte[] value) {
if (value == null) {
this.map.remove(key);
map.remove(key);
} else {
this.map.put(key, value);
map.put(key, value);
}
}

Expand All @@ -110,7 +108,7 @@ public void putAll(final List<KeyValue<Bytes, byte[]>> entries) {

@Override
public byte[] delete(final Bytes key) {
return this.map.remove(key);
return map.remove(key);
}

@Override
Expand All @@ -125,19 +123,19 @@ public KeyValueIterator<Bytes, byte[]> range(final Bytes from, final Bytes to) {

return new DelegatingPeekingKeyValueIterator<>(
name,
new InMemoryKeyValueIterator(this.map.subMap(from, true, to, true).entrySet().iterator()));
new InMemoryKeyValueIterator(map.subMap(from, true, to, true).entrySet().iterator()));
}

@Override
public KeyValueIterator<Bytes, byte[]> all() {
return new DelegatingPeekingKeyValueIterator<>(
name,
new InMemoryKeyValueIterator(this.map.entrySet().iterator()));
new InMemoryKeyValueIterator(map.entrySet().iterator()));
}

@Override
public long approximateNumEntries() {
return this.map.size();
return map.size();
}

@Override
Expand All @@ -147,8 +145,8 @@ public void flush() {

@Override
public void close() {
this.map.clear();
this.open = false;
map.clear();
open = false;
}

private static class InMemoryKeyValueIterator implements KeyValueIterator<Bytes, byte[]> {
Expand Down
Loading