root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

MINOR: Clean up ThreadCacheTest (#6485) 

Minor clean up ofThreadCacheTest
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Matthias J. Sax <mjsax@apache.org>
This commit is contained in:
Bill Bejeck 2019-03-22 09:27:58 -04:00 committed by GitHub
parent 1deb072f56
commit 1acae2a67c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 22 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
streams/src/test/java/org/apache/kafka/streams/state/internals/ThreadCacheTest.java
View File

@ -146,7 +146,7 @@ public class ThreadCacheTest {
} }
@Test @Test
public void evict() throws IOException { public void evict() {
final List<KeyValue<String, String>> received = new ArrayList<>(); final List<KeyValue<String, String>> received = new ArrayList<>();
final List<KeyValue<String, String>> expected = Collections.singletonList( final List<KeyValue<String, String>> expected = Collections.singletonList(
new KeyValue<>("K1", "V1")); new KeyValue<>("K1", "V1"));
@ -161,14 +161,10 @@ public class ThreadCacheTest {
final ThreadCache cache = new ThreadCache(logContext, final ThreadCache cache = new ThreadCache(logContext,
memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""), memoryCacheEntrySize(kv.key.getBytes(), kv.value.getBytes(), ""),
new MockStreamsMetrics(new Metrics())); new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, dirty -> {
@Override for (final ThreadCache.DirtyEntry dirtyEntry : dirty) {
public void apply(final List<ThreadCache.DirtyEntry> dirty) { received.add(new KeyValue<>(dirtyEntry.key().toString(), new String(dirtyEntry.newValue())));
for (final ThreadCache.DirtyEntry dirtyEntry : dirty) {
received.add(new KeyValue<>(dirtyEntry.key().toString(), new String(dirtyEntry.newValue())));
}
} }
}); });
for (final KeyValue<String, String> kvToInsert : toInsert) { for (final KeyValue<String, String> kvToInsert : toInsert) {
@ -200,12 +196,7 @@ public class ThreadCacheTest {
final Bytes key = Bytes.wrap(new byte[]{0}); final Bytes key = Bytes.wrap(new byte[]{0});
final ThreadCache cache = new ThreadCache(logContext, 10000L, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, 10000L, new MockStreamsMetrics(new Metrics()));
final List<ThreadCache.DirtyEntry> received = new ArrayList<>(); final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, received::addAll);
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
received.addAll(dirty);
}
});
cache.put(namespace, key, dirtyEntry(key.get())); cache.put(namespace, key, dirtyEntry(key.get()));
assertEquals(key.get(), cache.delete(namespace, key).value()); assertEquals(key.get(), cache.delete(namespace, key).value());
@ -298,12 +289,8 @@ public class ThreadCacheTest {
public void shouldSkipEntriesWhereValueHasBeenEvictedFromCache() { public void shouldSkipEntriesWhereValueHasBeenEvictedFromCache() {
final int entrySize = memoryCacheEntrySize(new byte[1], new byte[1], ""); final int entrySize = memoryCacheEntrySize(new byte[1], new byte[1], "");
final ThreadCache cache = new ThreadCache(logContext, entrySize * 5, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, entrySize * 5, new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, dirty -> { });
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
}
});
final byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}}; final byte[][] bytes = {{0}, {1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}};
for (int i = 0; i < 5; i++) { for (int i = 0; i < 5; i++) {
cache.put(namespace, Bytes.wrap(bytes[i]), dirtyEntry(bytes[i])); cache.put(namespace, Bytes.wrap(bytes[i]), dirtyEntry(bytes[i]));
@ -322,12 +309,9 @@ public class ThreadCacheTest {
public void shouldFlushDirtyEntriesForNamespace() { public void shouldFlushDirtyEntriesForNamespace() {
final ThreadCache cache = new ThreadCache(logContext, 100000, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, 100000, new MockStreamsMetrics(new Metrics()));
final List<byte[]> received = new ArrayList<>(); final List<byte[]> received = new ArrayList<>();
cache.addDirtyEntryFlushListener(namespace1, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace1, dirty -> {
@Override for (final ThreadCache.DirtyEntry dirtyEntry : dirty) {
public void apply(final List<ThreadCache.DirtyEntry> dirty) { received.add(dirtyEntry.key().get());
for (final ThreadCache.DirtyEntry dirtyEntry : dirty) {
received.add(dirtyEntry.key().get());
}
} }
}); });
final List<byte[]> expected = Arrays.asList(new byte[]{0}, new byte[]{1}, new byte[]{2}); final List<byte[]> expected = Arrays.asList(new byte[]{0}, new byte[]{1}, new byte[]{2});
@ -344,12 +328,9 @@ public class ThreadCacheTest {
public void shouldNotFlushCleanEntriesForNamespace() { public void shouldNotFlushCleanEntriesForNamespace() {
final ThreadCache cache = new ThreadCache(logContext, 100000, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, 100000, new MockStreamsMetrics(new Metrics()));
final List<byte[]> received = new ArrayList<>(); final List<byte[]> received = new ArrayList<>();
cache.addDirtyEntryFlushListener(namespace1, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace1, dirty -> {
@Override for (final ThreadCache.DirtyEntry dirtyEntry : dirty) {
public void apply(final List<ThreadCache.DirtyEntry> dirty) { received.add(dirtyEntry.key().get());
for (final ThreadCache.DirtyEntry dirtyEntry : dirty) {
received.add(dirtyEntry.key().get());
}
} }
}); });
final List<byte[]> toInsert = Arrays.asList(new byte[]{0}, new byte[]{1}, new byte[]{2}); final List<byte[]> toInsert = Arrays.asList(new byte[]{0}, new byte[]{1}, new byte[]{2});
@ -366,12 +347,7 @@ public class ThreadCacheTest {
private void shouldEvictImmediatelyIfCacheSizeIsZeroOrVerySmall(final ThreadCache cache) { private void shouldEvictImmediatelyIfCacheSizeIsZeroOrVerySmall(final ThreadCache cache) {
final List<ThreadCache.DirtyEntry> received = new ArrayList<>(); final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, received::addAll);
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
received.addAll(dirty);
}
});
cache.put(namespace, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{0})); cache.put(namespace, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{0}));
assertEquals(1, received.size()); assertEquals(1, received.size());
@ -396,12 +372,7 @@ public class ThreadCacheTest {
public void shouldEvictAfterPutAll() { public void shouldEvictAfterPutAll() {
final List<ThreadCache.DirtyEntry> received = new ArrayList<>(); final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
final ThreadCache cache = new ThreadCache(logContext, 1, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, 1, new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, received::addAll);
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
received.addAll(dirty);
}
});
cache.putAll(namespace, Arrays.asList(KeyValue.pair(Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5})), cache.putAll(namespace, Arrays.asList(KeyValue.pair(Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5})),
KeyValue.pair(Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6})))); KeyValue.pair(Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6}))));
@ -425,12 +396,7 @@ public class ThreadCacheTest {
public void shouldNotForwardCleanEntryOnEviction() { public void shouldNotForwardCleanEntryOnEviction() {
final ThreadCache cache = new ThreadCache(logContext, 0, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, 0, new MockStreamsMetrics(new Metrics()));
final List<ThreadCache.DirtyEntry> received = new ArrayList<>(); final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, received::addAll);
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
received.addAll(dirty);
}
});
cache.put(namespace, Bytes.wrap(new byte[]{1}), cleanEntry(new byte[]{0})); cache.put(namespace, Bytes.wrap(new byte[]{1}), cleanEntry(new byte[]{0}));
assertEquals(0, received.size()); assertEquals(0, received.size());
} }
@ -448,12 +414,7 @@ public class ThreadCacheTest {
public void shouldEvictAfterPutIfAbsent() { public void shouldEvictAfterPutIfAbsent() {
final List<ThreadCache.DirtyEntry> received = new ArrayList<>(); final List<ThreadCache.DirtyEntry> received = new ArrayList<>();
final ThreadCache cache = new ThreadCache(logContext, 1, new MockStreamsMetrics(new Metrics())); final ThreadCache cache = new ThreadCache(logContext, 1, new MockStreamsMetrics(new Metrics()));
cache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { cache.addDirtyEntryFlushListener(namespace, received::addAll);
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
received.addAll(dirty);
}
});
cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5})); cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[]{5}));
cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6})); cache.putIfAbsent(namespace, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[]{6}));
@ -468,26 +429,13 @@ public class ThreadCacheTest {
final int maxCacheSizeInBytes = 100; final int maxCacheSizeInBytes = 100;
final ThreadCache threadCache = new ThreadCache(logContext, maxCacheSizeInBytes, new MockStreamsMetrics(new Metrics())); final ThreadCache threadCache = new ThreadCache(logContext, maxCacheSizeInBytes, new MockStreamsMetrics(new Metrics()));
// trigger a put into another cache on eviction from "name" // trigger a put into another cache on eviction from "name"
threadCache.addDirtyEntryFlushListener(namespace, new ThreadCache.DirtyEntryFlushListener() { threadCache.addDirtyEntryFlushListener(namespace, dirty -> {
@Override // put an item into an empty cache when the total cache size
public void apply(final List<ThreadCache.DirtyEntry> dirty) { // is already > than maxCacheSizeBytes
// put an item into an empty cache when the total cache size threadCache.put(namespace1, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[2]));
// is already > than maxCacheSizeBytes
threadCache.put(namespace1, Bytes.wrap(new byte[]{0}), dirtyEntry(new byte[2]));
}
});
threadCache.addDirtyEntryFlushListener(namespace1, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
//
}
});
threadCache.addDirtyEntryFlushListener(namespace2, new ThreadCache.DirtyEntryFlushListener() {
@Override
public void apply(final List<ThreadCache.DirtyEntry> dirty) {
}
}); });
threadCache.addDirtyEntryFlushListener(namespace1, dirty -> { });
threadCache.addDirtyEntryFlushListener(namespace2, dirty -> { });
threadCache.put(namespace2, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[1])); threadCache.put(namespace2, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[1]));
threadCache.put(namespace, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[1])); threadCache.put(namespace, Bytes.wrap(new byte[]{1}), dirtyEntry(new byte[1]));