Skip to content

KAFKA-7080 and KAFKA-7222: Cleanup overlapping KIP changes Part 2 #5806

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

Closed
vvcephei wants to merge 2 commits into from
Closed

KAFKA-7080 and KAFKA-7222: Cleanup overlapping KIP changes Part 2 #5806

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cea137b
KAFKA-7080 and KAFKA-7222: Cleanup overlapping KIP changes Part 2
vvcephei Oct 16, 2018
257a1ce
don't ignore segments config in legacy mode
vvcephei Oct 16, 2018
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
7 changes: 3 additions & 4 deletions ...ams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,10 +192,9 @@ private <VR> StoreBuilder<WindowStore<K, VR>> materialize(final MaterializedInte

supplier = Stores.persistentWindowStore(
materialized.storeName(),
windows.maintainMs(),
windows.size(),
false,
windows.segmentInterval()
Duration.ofMillis(windows.maintainMs()),
Duration.ofMillis(windows.size()),
false
);
}
}
Expand Down
13 changes: 0 additions & 13 deletions streams/src/test/java/org/apache/kafka/streams/kstream/WindowsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,19 +40,6 @@ public long gracePeriodMs() {
}
}

@SuppressWarnings("deprecation") // specifically testing deprecated APIs
@Test
public void shouldSetNumberOfSegments() {
final int anySegmentSizeLargerThanOne = 5;
final TestWindows testWindow = new TestWindows();
final long maintainMs = testWindow.maintainMs();

assertEquals(
maintainMs / (anySegmentSizeLargerThanOne - 1),
testWindow.segments(anySegmentSizeLargerThanOne).segmentInterval()
);
}

@SuppressWarnings("deprecation") // specifically testing deprecated APIs
@Test
public void shouldSetWindowRetentionTime() {
Expand Down
7 changes: 3 additions & 4 deletions streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -473,10 +473,9 @@ private void processStreamWithWindowStore(final String topic) {
final StoreBuilder<WindowStore<Integer, byte[]>> storeBuilder = Stores.windowStoreBuilder(
Stores.persistentWindowStore(
"store",
AGGREGATE_WINDOW_SIZE * 3,
AGGREGATE_WINDOW_SIZE,
false,
60_000L
ofMillis(AGGREGATE_WINDOW_SIZE * 3),
ofMillis(AGGREGATE_WINDOW_SIZE),
false
),
INTEGER_SERDE,
BYTE_SERDE
Expand Down
10 changes: 3 additions & 7 deletions streams/src/test/java/org/apache/kafka/streams/state/StoresTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.kafka.streams.state.internals.RocksDBWindowStore;
import org.junit.Test;

import static java.time.Duration.ZERO;
import static java.time.Duration.ofMillis;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
Expand Down Expand Up @@ -55,12 +56,12 @@ public void shouldThrowIfILruMapStoreCapacityIsNegative() {

@Test(expected = NullPointerException.class)
public void shouldThrowIfIPersistentWindowStoreStoreNameIsNull() {
Stores.persistentWindowStore(null, 0L, 0L, false, 0L);
Stores.persistentWindowStore(null, ZERO, ZERO, false);
}

@Test(expected = IllegalArgumentException.class)
public void shouldThrowIfIPersistentWindowStoreRetentionPeriodIsNegative() {
Stores.persistentWindowStore("anyName", -1L, 0L, false, 0L);
Stores.persistentWindowStore("anyName", ofMillis(-1L), ZERO, false);
}

@Deprecated
Expand All @@ -74,11 +75,6 @@ public void shouldThrowIfIPersistentWindowStoreIfWindowSizeIsNegative() {
Stores.persistentWindowStore("anyName", ofMillis(0L), ofMillis(-1L), false);
}

@Test(expected = IllegalArgumentException.class)
public void shouldThrowIfIPersistentWindowStoreIfSegmentIntervalIsTooSmall() {
Stores.persistentWindowStore("anyName", 1L, 1L, false, -1L);
}

@Test(expected = NullPointerException.class)
public void shouldThrowIfIPersistentSessionStoreStoreNameIsNull() {
Stores.persistentSessionStore(null, ofMillis(0));
Expand Down
9 changes: 5 additions & 4 deletions streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBWindowStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import java.util.Map;
import java.util.Set;

import static java.time.Duration.ofMillis;
import static java.time.Instant.ofEpochMilli;
import static java.util.Objects.requireNonNull;
import static org.hamcrest.CoreMatchers.equalTo;
Expand All @@ -70,7 +71,7 @@ public class RocksDBWindowStoreTest {

private final int numSegments = 3;
private final long windowSize = 3L;
private final long segmentInterval = 600L;
private final long segmentInterval = 60_000L;
private final long retentionPeriod = segmentInterval * (numSegments - 1);
private final String windowName = "window";
private final Segments segments = new Segments(windowName, retentionPeriod, segmentInterval);
Expand Down Expand Up @@ -108,7 +109,7 @@ public <K1, V1> void send(final String topic,

private WindowStore<Integer, String> createWindowStore(final ProcessorContext context, final boolean retainDuplicates) {
final WindowStore<Integer, String> store = Stores.windowStoreBuilder(
Stores.persistentWindowStore(windowName, retentionPeriod, windowSize, retainDuplicates, segmentInterval),

@vvcephei vvcephei Oct 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@dguy Yes, this is where I've updated it.

@dguy dguy Oct 16, 2018

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.

I think that method is private?? I might have missed somerhing

@vvcephei vvcephei Oct 16, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

It was previously public. #5804 made it private, but did not remove this reference. This is why trunk fails to build right now.

I'm removing this reference and instead calling a public method.

Stores.persistentWindowStore(windowName, ofMillis(retentionPeriod), ofMillis(windowSize), retainDuplicates),
Serdes.Integer(),
Serdes.String()).build();

Expand Down Expand Up @@ -771,7 +772,7 @@ public void shouldFetchAndIterateOverExactKeys() {
final long retentionPeriod = 0x7a00000000000000L;

final WindowStore<String, String> windowStore = Stores.windowStoreBuilder(
Stores.persistentWindowStore(windowName, retentionPeriod, windowSize, true),
Stores.persistentWindowStore(windowName, ofMillis(retentionPeriod), ofMillis(windowSize), true),
Serdes.String(),
Serdes.String()).build();

Expand Down Expand Up @@ -848,7 +849,7 @@ public void shouldNoNullPointerWhenSerdeDoesNotHandleNull() {
@Test
public void shouldFetchAndIterateOverExactBinaryKeys() {
final WindowStore<Bytes, String> windowStore = Stores.windowStoreBuilder(
Stores.persistentWindowStore(windowName, 60_000L, 60_000L, true),
Stores.persistentWindowStore(windowName, ofMillis(60_000L), ofMillis(60_000L), true),
Serdes.Bytes(),
Serdes.String()).build();

Expand Down