Skip to content

MINOR: cleanup RocksDBStore tests #8510

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 4 commits into from
Apr 18, 2020
Merged

MINOR: cleanup RocksDBStore tests #8510

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
74859e1
use temp dir
ableegoldman Apr 18, 2020
6950056
remove unused import
ableegoldman Apr 18, 2020
1ff249b
undo static import
ableegoldman Apr 18, 2020
e338325
fix uo
ableegoldman Apr 18, 2020
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
118 changes: 47 additions & 71 deletions streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,8 +74,8 @@
import static org.hamcrest.Matchers.greaterThan;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.powermock.api.easymock.PowerMock.replay;
import static org.powermock.api.easymock.PowerMock.verify;

Expand Down Expand Up @@ -106,6 +106,11 @@ public void setUp() {
context.metrics().setRocksDBMetricsRecordingTrigger(new RocksDBMetricsRecordingTrigger());
}

@After
public void tearDown() {
rocksDBStore.close();
}

RocksDBStore getRocksDBStore() {
return new RocksDBStore(DB_NAME, METRICS_SCOPE);
}
Expand Down Expand Up @@ -137,62 +142,57 @@ private InternalMockProcessorContext getProcessorContext(final RecordingLevel re
return getProcessorContext(streamsProps);
}

@After
public void tearDown() {
rocksDBStore.close();
}

@Test
public void shouldAddStatisticsToInjectedMetricsRecorderWhenRecordingLevelIsDebug() {
final RocksDBStore store = getRocksDBStoreWithRocksDBMetricsRecorder();
final InternalMockProcessorContext mockContext = getProcessorContext(RecordingLevel.DEBUG);
context = getProcessorContext(RecordingLevel.DEBUG);

@guozhangwang guozhangwang Apr 18, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: for line 147 could we use the rocksDBStore as well?

@ableegoldman ableegoldman Apr 18, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I left it out intentionally because we always cause #close on this rocksDBStore and this causes this test to fail.

It's just because of the mock, so we could instead add some expect... lines at the end of the test, but it seemed cleaner to just use a local RocksDBStore here

reset(metricsRecorder);
metricsRecorder.addStatistics(
eq(DB_NAME),
anyObject(Statistics.class)
);
replay(metricsRecorder);

store.openDB(mockContext);
store.openDB(context);

verify(metricsRecorder);
}

@Test
public void shouldNotAddStatisticsToInjectedMetricsRecorderWhenRecordingLevelIsInfo() {
final RocksDBStore store = getRocksDBStoreWithRocksDBMetricsRecorder();
final InternalMockProcessorContext mockContext = getProcessorContext(RecordingLevel.INFO);
rocksDBStore= getRocksDBStoreWithRocksDBMetricsRecorder();
context = getProcessorContext(RecordingLevel.INFO);
reset(metricsRecorder);
replay(metricsRecorder);

store.openDB(mockContext);
rocksDBStore.openDB(context);

verify(metricsRecorder);
}

@Test
public void shouldRemoveStatisticsFromInjectedMetricsRecorderOnCloseWhenRecordingLevelIsDebug() {
final RocksDBStore store = getRocksDBStoreWithRocksDBMetricsRecorder();
final InternalMockProcessorContext mockContext = getProcessorContext(RecordingLevel.DEBUG);
store.openDB(mockContext);
rocksDBStore = getRocksDBStoreWithRocksDBMetricsRecorder();
context = getProcessorContext(RecordingLevel.DEBUG);
rocksDBStore.openDB(context);
reset(metricsRecorder);
metricsRecorder.removeStatistics(DB_NAME);
replay(metricsRecorder);

store.close();
rocksDBStore.close();

verify(metricsRecorder);
}

@Test
public void shouldNotRemoveStatisticsFromInjectedMetricsRecorderOnCloseWhenRecordingLevelIsInfo() {
final RocksDBStore store = getRocksDBStoreWithRocksDBMetricsRecorder();
final InternalMockProcessorContext mockContext = getProcessorContext(RecordingLevel.INFO);
store.openDB(mockContext);
rocksDBStore = getRocksDBStoreWithRocksDBMetricsRecorder();
context = getProcessorContext(RecordingLevel.INFO);
rocksDBStore.openDB(context);
reset(metricsRecorder);
replay(metricsRecorder);

store.close();
rocksDBStore.close();

verify(metricsRecorder);
}
Expand All @@ -211,26 +211,22 @@ public void close(final String storeName, final Options options) {

@Test
public void shouldNotAddStatisticsToInjectedMetricsRecorderWhenUserProvidesStatistics() {
final RocksDBStore store = getRocksDBStoreWithRocksDBMetricsRecorder();
final InternalMockProcessorContext mockContext =
getProcessorContext(RecordingLevel.DEBUG, RocksDBConfigSetterWithUserProvidedStatistics.class);
rocksDBStore = getRocksDBStoreWithRocksDBMetricsRecorder();
context = getProcessorContext(RecordingLevel.DEBUG, RocksDBConfigSetterWithUserProvidedStatistics.class);
replay(metricsRecorder);

store.openDB(mockContext);
rocksDBStore.openDB(context);
verify(metricsRecorder);
}

@Test
public void shouldNotRemoveStatisticsFromInjectedMetricsRecorderOnCloseWhenUserProvidesStatistics() {
final RocksDBStore store = getRocksDBStoreWithRocksDBMetricsRecorder();
final InternalMockProcessorContext mockContext =
getProcessorContext(RecordingLevel.DEBUG, RocksDBConfigSetterWithUserProvidedStatistics.class);
store.openDB(mockContext);
rocksDBStore = getRocksDBStoreWithRocksDBMetricsRecorder();
context = getProcessorContext(RecordingLevel.DEBUG, RocksDBConfigSetterWithUserProvidedStatistics.class);
rocksDBStore.openDB(context);
reset(metricsRecorder);
replay(metricsRecorder);

store.close();

verify(metricsRecorder);
}

Expand Down Expand Up @@ -290,12 +286,7 @@ public void shouldThrowProcessorStateExceptionOnOpeningReadOnlyDir() {

assertTrue(tmpDir.setReadOnly());

try {
rocksDBStore.openDB(tmpContext);
fail("Should have thrown ProcessorStateException");
} catch (final ProcessorStateException e) {
// this is good, do nothing
}
assertThrows(ProcessorStateException.class, () -> rocksDBStore.openDB(tmpContext));
}

@Test
Expand Down Expand Up @@ -503,56 +494,41 @@ public void shouldRestoreThenDeleteOnRestoreAll() {
@Test
public void shouldThrowNullPointerExceptionOnNullPut() {
rocksDBStore.init(context, rocksDBStore);
try {
rocksDBStore.put(null, stringSerializer.serialize(null, "someVal"));
fail("Should have thrown NullPointerException on null put()");
} catch (final NullPointerException e) {
// this is good
}
assertThrows(
NullPointerException.class,
() -> rocksDBStore.put(null, stringSerializer.serialize(null, "someVal")));
}

@Test
public void shouldThrowNullPointerExceptionOnNullPutAll() {
rocksDBStore.init(context, rocksDBStore);
try {
rocksDBStore.put(null, stringSerializer.serialize(null, "someVal"));
fail("Should have thrown NullPointerException on null put()");
} catch (final NullPointerException e) {
// this is good
}
assertThrows(
NullPointerException.class,
() -> rocksDBStore.put(null, stringSerializer.serialize(null, "someVal")));
}

@Test
public void shouldThrowNullPointerExceptionOnNullGet() {
rocksDBStore.init(context, rocksDBStore);
try {
rocksDBStore.get(null);
fail("Should have thrown NullPointerException on null get()");
} catch (final NullPointerException e) {
// this is good
}
assertThrows(
NullPointerException.class,
() -> rocksDBStore.get(null));
}

@Test
public void shouldThrowNullPointerExceptionOnDelete() {
rocksDBStore.init(context, rocksDBStore);
try {
rocksDBStore.delete(null);
fail("Should have thrown NullPointerException on deleting null key");
} catch (final NullPointerException e) {
// this is good
}
assertThrows(
NullPointerException.class,
() -> rocksDBStore.delete(null));
}

@Test
public void shouldThrowNullPointerExceptionOnRange() {
rocksDBStore.init(context, rocksDBStore);
try {
rocksDBStore.range(null, new Bytes(stringSerializer.serialize(null, "2")));
fail("Should have thrown NullPointerException on null range key");
} catch (final NullPointerException e) {
// this is good
}
assertThrows(
NullPointerException.class,
() -> rocksDBStore.range(null, new Bytes(stringSerializer.serialize(null, "2"))));
}

@Test(expected = ProcessorStateException.class)
Expand All @@ -567,10 +543,8 @@ public void shouldThrowProcessorStateExceptionOnPutDeletedDir() throws IOExcepti

@Test
public void shouldHandleToggleOfEnablingBloomFilters() {

final Properties props = StreamsTestUtils.getStreamsConfig();
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, TestingBloomFilterRocksDBConfigSetter.class);
rocksDBStore = getRocksDBStore();
dir = TestUtils.tempDirectory();
context = new InternalMockProcessorContext(dir,
Serdes.String(),
Expand Down Expand Up @@ -616,20 +590,22 @@ public void shouldHandleToggleOfEnablingBloomFilters() {

@Test
public void shouldVerifyThatMetricsGetMeasurementsFromRocksDB() {
final TaskId taskId = new TaskId(0, 0);

final RocksDBMetricsRecordingTrigger rocksDBMetricsRecordingTrigger = new RocksDBMetricsRecordingTrigger();
final Metrics metrics = new Metrics(new MetricConfig().recordLevel(RecordingLevel.DEBUG));
final StreamsMetricsImpl streamsMetrics =
new StreamsMetricsImpl(metrics, "test-application", StreamsConfig.METRICS_LATEST);
streamsMetrics.setRocksDBMetricsRecordingTrigger(rocksDBMetricsRecordingTrigger);
final ProcessorContext<Object, Object> context = EasyMock.niceMock(ProcessorContext.class);

context = EasyMock.niceMock(InternalMockProcessorContext.class);
EasyMock.expect(context.metrics()).andStubReturn(streamsMetrics);
final TaskId taskId = new TaskId(0, 0);
EasyMock.expect(context.taskId()).andStubReturn(taskId);
EasyMock.expect(context.appConfigs())
.andStubReturn(new StreamsConfig(StreamsTestUtils.getStreamsConfig()).originals());
EasyMock.expect(context.taskId()).andStubReturn(taskId);
EasyMock.expect(context.stateDir()).andStubReturn(dir);
EasyMock.replay(context);
final RocksDBStore rocksDBStore = new RocksDBStore(DB_NAME, METRICS_SCOPE);

rocksDBStore.init(context, rocksDBStore);
final byte[] key = "hello".getBytes();
final byte[] value = "world".getBytes();
Expand Down
4 changes: 0 additions & 4 deletions ...s/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,13 +71,10 @@ public void shouldOpenExistingStoreInRegularMode() throws Exception {

// re-open store
final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
rocksDBStore = getRocksDBStore();
rocksDBStore.init(context, rocksDBStore);
assertThat(appender.getMessages(), hasItem("Opening store " + DB_NAME + " in regular mode"));
LogCaptureAppender.unregister(appender);

rocksDBStore.close();

// verify store
final DBOptions dbOptions = new DBOptions();
final ColumnFamilyOptions columnFamilyOptions = new ColumnFamilyOptions();
Expand Down Expand Up @@ -201,7 +198,6 @@ public void shouldMigrateDataFromDefaultToTimestampColumnFamily() throws Excepti
// approx: 0 entries on old CF, 3 in new CF
assertThat(rocksDBStore.approximateNumEntries(), is(3L));


iteratorsShouldNotMigrateData();
assertThat(rocksDBStore.approximateNumEntries(), is(3L));

Expand Down