tchekai704
/
src-kafka
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

KAFKA-14133: Migrato SessionCacheFlushListenerTest, TimestampedCacheFlushListenerTest and TimestampedTupleForwarderTest to Mockito (#14205) 

Reviewers: Divij Vaidya <diviv@amazon.com>
pull/14472/head
Christo Lolov 1 year ago committed by GitHub
parent
ee27773549
commit
1a15cd708a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 42 additions and 53 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 22
      streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionCacheFlushListenerTest.java
  2. 32
      streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimestampedCacheFlushListenerTest.java
  3. 41
      streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimestampedTupleForwarderTest.java

22
streams/src/test/java/org/apache/kafka/streams/kstream/internals/SessionCacheFlushListenerTest.java
Unescape View File

@ -20,27 +20,25 @@ import org.apache.kafka.streams.kstream.Windowed; @@ -20,27 +20,25 @@ import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.junit.MockitoJUnitRunner;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@RunWith(MockitoJUnitRunner.StrictStubs.class)
public class SessionCacheFlushListenerTest {
@Test
public void shouldForwardKeyNewValueOldValueAndTimestamp() {
@SuppressWarnings("unchecked")
final InternalProcessorContext<Windowed<String>, Change<String>> context = mock(InternalProcessorContext.class);
expect(context.currentNode()).andReturn(null).anyTimes();
context.setCurrentNode(null);
context.setCurrentNode(null);
context.forward(
doNothing().when(context).forward(
new Record<>(
new Windowed<>("key", new SessionWindow(21L, 73L)),
new Change<>("newValue", "oldValue"),
73L));
expectLastCall();
replay(context);
new SessionCacheFlushListener<>(context).apply(
new Record<>(
@ -48,6 +46,6 @@ public class SessionCacheFlushListenerTest { @@ -48,6 +46,6 @@ public class SessionCacheFlushListenerTest {
new Change<>("newValue", "oldValue"),
42L));
verify(context);
verify(context, times(2)).setCurrentNode(null);
}
}

32
streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimestampedCacheFlushListenerTest.java
Unescape View File

@ -20,28 +20,26 @@ import org.apache.kafka.streams.processor.api.Record; @@ -20,28 +20,26 @@ import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.junit.MockitoJUnitRunner;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@RunWith(MockitoJUnitRunner.StrictStubs.class)
public class TimestampedCacheFlushListenerTest {
@Test
public void shouldForwardValueTimestampIfNewValueExists() {
@SuppressWarnings("unchecked")
final InternalProcessorContext<String, Change<String>> context = mock(InternalProcessorContext.class);
expect(context.currentNode()).andReturn(null).anyTimes();
context.setCurrentNode(null);
context.setCurrentNode(null);
context.forward(
doNothing().when(context).forward(
new Record<>(
"key",
new Change<>("newValue", "oldValue"),
42L));
expectLastCall();
replay(context);
new TimestampedCacheFlushListener<>(context).apply(
new Record<>(
@ -51,22 +49,18 @@ public class TimestampedCacheFlushListenerTest { @@ -51,22 +49,18 @@ public class TimestampedCacheFlushListenerTest {
ValueAndTimestamp.make("oldValue", 21L)),
73L));
verify(context);
verify(context, times(2)).setCurrentNode(null);
}
@Test
public void shouldForwardParameterTimestampIfNewValueIsNull() {
@SuppressWarnings("unchecked")
final InternalProcessorContext<String, Change<String>> context = mock(InternalProcessorContext.class);
expect(context.currentNode()).andReturn(null).anyTimes();
context.setCurrentNode(null);
context.setCurrentNode(null);
context.forward(
doNothing().when(context).forward(
new Record<>(
"key",
new Change<>(null, "oldValue"),
73L));
expectLastCall();
replay(context);
new TimestampedCacheFlushListener<>(context).apply(
new Record<>(
@ -74,6 +68,6 @@ public class TimestampedCacheFlushListenerTest { @@ -74,6 +68,6 @@ public class TimestampedCacheFlushListenerTest {
new Change<>(null, ValueAndTimestamp.make("oldValue", 21L)),
73L));
verify(context);
verify(context, times(2)).setCurrentNode(null);
}
}

41
streams/src/test/java/org/apache/kafka/streams/kstream/internals/TimestampedTupleForwarderTest.java
Unescape View File

@ -22,13 +22,14 @@ import org.apache.kafka.streams.processor.internals.InternalProcessorContext; @@ -22,13 +22,14 @@ import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.junit.MockitoJUnitRunner;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.mock;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.verify;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.StrictStubs.class)
public class TimestampedTupleForwarderTest {
@Test
@ -38,11 +39,12 @@ public class TimestampedTupleForwarderTest { @@ -38,11 +39,12 @@ public class TimestampedTupleForwarderTest {
}
private void setFlushListener(final boolean sendOldValues) {
@SuppressWarnings("unchecked")
final WrappedStateStore<StateStore, Object, ValueAndTimestamp<Object>> store = mock(WrappedStateStore.class);
@SuppressWarnings("unchecked")
final TimestampedCacheFlushListener<Object, Object> flushListener = mock(TimestampedCacheFlushListener.class);
expect(store.setFlushListener(flushListener, sendOldValues)).andReturn(false);
replay(store);
when(store.setFlushListener(flushListener, sendOldValues)).thenReturn(false);
new TimestampedTupleForwarder<>(
store,
@ -50,8 +52,6 @@ public class TimestampedTupleForwarderTest { @@ -50,8 +52,6 @@ public class TimestampedTupleForwarderTest {
flushListener,
sendOldValues
);
verify(store);
}
@Test
@ -61,19 +61,19 @@ public class TimestampedTupleForwarderTest { @@ -61,19 +61,19 @@ public class TimestampedTupleForwarderTest {
}
private void shouldForwardRecordsIfWrappedStateStoreDoesNotCache(final boolean sendOldValues) {
@SuppressWarnings("unchecked")
final WrappedStateStore<StateStore, String, String> store = mock(WrappedStateStore.class);
@SuppressWarnings("unchecked")
final InternalProcessorContext<String, Change<String>> context = mock(InternalProcessorContext.class);
expect(store.setFlushListener(null, sendOldValues)).andReturn(false);
when(store.setFlushListener(null, sendOldValues)).thenReturn(false);
if (sendOldValues) {
context.forward(new Record<>("key1", new Change<>("newValue1", "oldValue1", true), 0L));
context.forward(new Record<>("key2", new Change<>("newValue2", "oldValue2", false), 42L));
doNothing().when(context).forward(new Record<>("key1", new Change<>("newValue1", "oldValue1", true), 0L));
doNothing().when(context).forward(new Record<>("key2", new Change<>("newValue2", "oldValue2", false), 42L));
} else {
context.forward(new Record<>("key1", new Change<>("newValue1", null, true), 0L));
context.forward(new Record<>("key2", new Change<>("newValue2", null, false), 42L));
doNothing().when(context).forward(new Record<>("key1", new Change<>("newValue1", null, true), 0L));
doNothing().when(context).forward(new Record<>("key2", new Change<>("newValue2", null, false), 42L));
}
expectLastCall();
replay(store, context);
final TimestampedTupleForwarder<String, String> forwarder =
new TimestampedTupleForwarder<>(
@ -84,17 +84,16 @@ public class TimestampedTupleForwarderTest { @@ -84,17 +84,16 @@ public class TimestampedTupleForwarderTest {
);
forwarder.maybeForward(new Record<>("key1", new Change<>("newValue1", "oldValue1", true), 0L));
forwarder.maybeForward(new Record<>("key2", new Change<>("newValue2", "oldValue2", false), 42L));
verify(store, context);
}
@Test
public void shouldNotForwardRecordsIfWrappedStateStoreDoesCache() {
@SuppressWarnings("unchecked")
final WrappedStateStore<StateStore, String, String> store = mock(WrappedStateStore.class);
@SuppressWarnings("unchecked")
final InternalProcessorContext<String, Change<String>> context = mock(InternalProcessorContext.class);
expect(store.setFlushListener(null, false)).andReturn(true);
replay(store, context);
when(store.setFlushListener(null, false)).thenReturn(true);
final TimestampedTupleForwarder<String, String> forwarder =
new TimestampedTupleForwarder<>(
@ -105,7 +104,5 @@ public class TimestampedTupleForwarderTest { @@ -105,7 +104,5 @@ public class TimestampedTupleForwarderTest {
);
forwarder.maybeForward(new Record<>("key", new Change<>("newValue", "oldValue", true), 0L));
forwarder.maybeForward(new Record<>("key", new Change<>("newValue", "oldValue", true), 42L));
verify(store, context);
}
}

Write Preview
Loading…
Cancel
Save