KAFKA-8025: Fix flaky RocksDB test (#8126)

Reviewers: Bill Bejeck <bill@confluent.io>

streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest.java

@@ -18,7 +18,6 @@ package org.apache.kafka.streams.state.internals;
 
 import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender;
 import org.easymock.EasyMockRunner;
-import org.easymock.Mock;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.rocksdb.AbstractCompactionFilter;
@@ -55,6 +54,7 @@ import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 
+import static org.easymock.EasyMock.mock;
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.reset;
 import static org.easymock.EasyMock.verify;
@@ -85,11 +85,6 @@ public class RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest {
         }
     };
 
-    @Mock
-    private DBOptions dbOptions;
-    @Mock
-    private ColumnFamilyOptions columnFamilyOptions;
-
     @Test
     public void shouldOverwriteAllOptionsMethods() throws Exception {
         for (final Method method : Options.class.getMethods()) {
@@ -112,47 +107,16 @@ public class RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest {
         }
     }
 
-    @Test
-    public void shouldWarnThanMethodCompactionOptionsFIFOSetTtlWillBeRemoved() {
-        final RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter optionsFacadeDbOptions
-            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(dbOptions, new ColumnFamilyOptions());
-
-        final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
-        optionsFacadeDbOptions.setCompactionOptionsFIFO(new CompactionOptionsFIFO());
-        LogCaptureAppender.unregister(appender);
-
-        assertThat(appender.getMessages(), hasItem(""
-            + "RocksDB's version will be bumped to version 6+ via KAFKA-8897 in a future release. "
-            + "If you use `org.rocksdb.CompactionOptionsFIFO#setTtl(long)` or `#ttl()` you will need to rewrite "
-            + "your code after KAFKA-8897 is resolved and set TTL via `org.rocksdb.Options` "
-            + "(or `org.rocksdb.ColumnFamilyOptions`)."));
-    }
-
-    @Test
-    public void shouldWarnThanMethodCompactionOptionsFIFOTtlWillBeRemoved() {
-        final RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter optionsFacadeDbOptions
-            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(dbOptions, new ColumnFamilyOptions());
-
-        final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
-        optionsFacadeDbOptions.compactionOptionsFIFO();
-        LogCaptureAppender.unregister(appender);
-
-        assertThat(appender.getMessages(), hasItem(""
-            + "RocksDB's version will be bumped to version 6+ via KAFKA-8897 in a future release. "
-            + "If you use `org.rocksdb.CompactionOptionsFIFO#setTtl(long)` or `#ttl()` you will need to rewrite "
-            + "your code after KAFKA-8897 is resolved and set TTL via `org.rocksdb.Options` "
-            + "(or `org.rocksdb.ColumnFamilyOptions`)."));
-    }
-
     private void verifyDBOptionsMethodCall(final Method method) throws Exception {
+        final DBOptions mockedDbOptions = mock(DBOptions.class);
         final RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter optionsFacadeDbOptions
-            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(dbOptions, new ColumnFamilyOptions());
+            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(mockedDbOptions, new ColumnFamilyOptions());
 
         final Object[] parameters = getDBOptionsParameters(method.getParameterTypes());
 
         try {
-            reset(dbOptions);
-            replay(dbOptions);
+            reset(mockedDbOptions);
+            replay(mockedDbOptions);
             method.invoke(optionsFacadeDbOptions, parameters);
             verify();
             fail("Should have called DBOptions." + method.getName() + "()");
@@ -231,14 +195,15 @@ public class RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest {
     }
 
     private void verifyColumnFamilyOptionsMethodCall(final Method method) throws Exception {
+        final ColumnFamilyOptions mockedColumnFamilyOptions = mock(ColumnFamilyOptions.class);
         final RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter optionsFacadeColumnFamilyOptions
-            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(new DBOptions(), columnFamilyOptions);
+            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(new DBOptions(), mockedColumnFamilyOptions);
 
         final Object[] parameters = getColumnFamilyOptionsParameters(method.getParameterTypes());
 
         try {
-            reset(columnFamilyOptions);
-            replay(columnFamilyOptions);
+            reset(mockedColumnFamilyOptions);
+            replay(mockedColumnFamilyOptions);
             method.invoke(optionsFacadeColumnFamilyOptions, parameters);
             verify();
             fail("Should have called ColumnFamilyOptions." + method.getName() + "()");
@@ -320,4 +285,38 @@ public class RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest {
 
         return parameters;
     }
+
+    @Test
+    public void shouldWarnThanMethodCompactionOptionsFIFOSetTtlWillBeRemoved() {
+        final DBOptions mockedDbOptions = mock(DBOptions.class);
+        final RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter optionsFacadeDbOptions
+            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(mockedDbOptions, new ColumnFamilyOptions());
+
+        final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
+        optionsFacadeDbOptions.setCompactionOptionsFIFO(new CompactionOptionsFIFO());
+        LogCaptureAppender.unregister(appender);
+
+        assertThat(appender.getMessages(), hasItem(""
+            + "RocksDB's version will be bumped to version 6+ via KAFKA-8897 in a future release. "
+            + "If you use `org.rocksdb.CompactionOptionsFIFO#setTtl(long)` or `#ttl()` you will need to rewrite "
+            + "your code after KAFKA-8897 is resolved and set TTL via `org.rocksdb.Options` "
+            + "(or `org.rocksdb.ColumnFamilyOptions`)."));
+    }
+
+    @Test
+    public void shouldWarnThanMethodCompactionOptionsFIFOTtlWillBeRemoved() {
+        final DBOptions mockedDbOptions = mock(DBOptions.class);
+        final RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter optionsFacadeDbOptions
+            = new RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapter(mockedDbOptions, new ColumnFamilyOptions());
+
+        final LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
+        optionsFacadeDbOptions.compactionOptionsFIFO();
+        LogCaptureAppender.unregister(appender);
+
+        assertThat(appender.getMessages(), hasItem(""
+            + "RocksDB's version will be bumped to version 6+ via KAFKA-8897 in a future release. "
+            + "If you use `org.rocksdb.CompactionOptionsFIFO#setTtl(long)` or `#ttl()` you will need to rewrite "
+            + "your code after KAFKA-8897 is resolved and set TTL via `org.rocksdb.Options` "
+            + "(or `org.rocksdb.ColumnFamilyOptions`)."));
+    }
 }