You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2018/08/24 00:53:16 UTC
[kafka] branch trunk updated: MINOR: restructure Windows to favor
immutable implementation (#5536)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 1dd85a3 MINOR: restructure Windows to favor immutable implementation (#5536)
1dd85a3 is described below
commit 1dd85a328f437643a5c135f242c644f47c3767e4
Author: John Roesler <vv...@users.noreply.github.com>
AuthorDate: Thu Aug 23 19:53:10 2018 -0500
MINOR: restructure Windows to favor immutable implementation (#5536)
Update to KIP-328.
Reviewers: Matthias J. Sax <ma...@confluent.io>, Guozhang Wang <gu...@confluent.io>, Ted Yu <yu...@gmail.com>, Kamal Chandraprakash <ka...@gmail.com>
---
.../apache/kafka/streams/kstream/JoinWindows.java | 84 +++++++++--
.../kafka/streams/kstream/SessionWindows.java | 5 +-
.../apache/kafka/streams/kstream/TimeWindows.java | 83 +++++++++--
.../kafka/streams/kstream/UnlimitedWindows.java | 22 ++-
.../org/apache/kafka/streams/kstream/Windows.java | 77 ++---------
.../kstream/internals/WindowingDefaults.java | 23 ++++
.../org/apache/kafka/streams/EqualityCheck.java | 153 +++++++++++++++++++++
.../KStreamAggregationIntegrationTest.java | 90 ++++++++++++
.../kafka/streams/kstream/JoinWindowsTest.java | 69 +++-------
.../kafka/streams/kstream/SessionWindowsTest.java | 34 ++---
.../kafka/streams/kstream/TimeWindowsTest.java | 52 ++-----
.../streams/kstream/UnlimitedWindowsTest.java | 22 +--
.../apache/kafka/streams/kstream/WindowsTest.java | 19 +--
13 files changed, 481 insertions(+), 252 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
index f91182e..5e742e1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
@@ -18,9 +18,12 @@ package org.apache.kafka.streams.kstream;
import org.apache.kafka.streams.processor.TimestampExtractor;
+import java.time.Duration;
import java.util.Map;
import java.util.Objects;
+import static org.apache.kafka.streams.kstream.internals.WindowingDefaults.DEFAULT_RETENTION_MS;
+
/**
* The window specifications used for joins.
* <p>
@@ -65,17 +68,39 @@ import java.util.Objects;
*/
public final class JoinWindows extends Windows<Window> {
+ private final long maintainDurationMs;
+
/** Maximum time difference for tuples that are before the join tuple. */
public final long beforeMs;
/** Maximum time difference for tuples that are after the join tuple. */
public final long afterMs;
- private JoinWindows(final long beforeMs, final long afterMs) {
+ private final Duration grace;
+
+ private JoinWindows(final long beforeMs, final long afterMs, final Duration grace, final long maintainDurationMs) {
+ if (beforeMs + afterMs < 0) {
+ throw new IllegalArgumentException("Window interval (ie, beforeMs+afterMs) must not be negative.");
+ }
+ this.afterMs = afterMs;
+ this.beforeMs = beforeMs;
+ this.grace = grace;
+ this.maintainDurationMs = maintainDurationMs;
+ }
+
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
+ private JoinWindows(final long beforeMs,
+ final long afterMs,
+ final Duration grace,
+ final long maintainDurationMs,
+ final int segments) {
+ super(segments);
if (beforeMs + afterMs < 0) {
throw new IllegalArgumentException("Window interval (ie, beforeMs+afterMs) must not be negative.");
}
this.afterMs = afterMs;
this.beforeMs = beforeMs;
+ this.grace = grace;
+ this.maintainDurationMs = maintainDurationMs;
}
/**
@@ -87,7 +112,8 @@ public final class JoinWindows extends Windows<Window> {
* @throws IllegalArgumentException if {@code timeDifferenceMs} is negative
*/
public static JoinWindows of(final long timeDifferenceMs) throws IllegalArgumentException {
- return new JoinWindows(timeDifferenceMs, timeDifferenceMs);
+ // This is a static factory method, so we initialize grace and retention to the defaults.
+ return new JoinWindows(timeDifferenceMs, timeDifferenceMs, null, DEFAULT_RETENTION_MS);
}
/**
@@ -100,8 +126,9 @@ public final class JoinWindows extends Windows<Window> {
* @param timeDifferenceMs relative window start time in milliseconds
* @throws IllegalArgumentException if the resulting window size is negative
*/
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
public JoinWindows before(final long timeDifferenceMs) throws IllegalArgumentException {
- return new JoinWindows(timeDifferenceMs, afterMs);
+ return new JoinWindows(timeDifferenceMs, afterMs, grace, maintainDurationMs, segments);
}
/**
@@ -114,8 +141,9 @@ public final class JoinWindows extends Windows<Window> {
* @param timeDifferenceMs relative window end time in milliseconds
* @throws IllegalArgumentException if the resulting window size is negative
*/
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
public JoinWindows after(final long timeDifferenceMs) throws IllegalArgumentException {
- return new JoinWindows(beforeMs, timeDifferenceMs);
+ return new JoinWindows(beforeMs, timeDifferenceMs, grace, maintainDurationMs, segments);
}
/**
@@ -134,10 +162,30 @@ public final class JoinWindows extends Windows<Window> {
return beforeMs + afterMs;
}
- @Override
+ /**
+ * Reject late events that arrive more than {@code millisAfterWindowEnd}
+ * after the end of its window.
+ *
+ * Lateness is defined as (stream_time - record_timestamp).
+ *
+ * @param millisAfterWindowEnd The grace period to admit late-arriving events to a window.
+ * @return this updated builder
+ */
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
public JoinWindows grace(final long millisAfterWindowEnd) {
- super.grace(millisAfterWindowEnd);
- return this;
+ if (millisAfterWindowEnd < 0) {
+ throw new IllegalArgumentException("Grace period must not be negative.");
+ }
+ return new JoinWindows(beforeMs, afterMs, Duration.ofMillis(millisAfterWindowEnd), maintainDurationMs, segments);
+ }
+
+ @SuppressWarnings("deprecation") // continuing to support Windows#maintainMs/segmentInterval in fallback mode
+ @Override
+ public long gracePeriodMs() {
+ // NOTE: in the future, when we remove maintainMs,
+ // we should default the grace period to 24h to maintain the default behavior,
+ // or we can default to (24h - size) if you want to be super accurate.
+ return grace != null ? grace.toMillis() : maintainMs() - size();
}
/**
@@ -146,14 +194,14 @@ public final class JoinWindows extends Windows<Window> {
* @throws IllegalArgumentException if {@code durationMs} is smaller than the window size
* @deprecated since 2.1. Use {@link JoinWindows#grace(long)} instead.
*/
+ @SuppressWarnings("deprecation")
@Override
@Deprecated
public JoinWindows until(final long durationMs) throws IllegalArgumentException {
if (durationMs < size()) {
throw new IllegalArgumentException("Window retention time (durationMs) cannot be smaller than the window size.");
}
- super.until(durationMs);
- return this;
+ return new JoinWindows(beforeMs, afterMs, grace, durationMs, segments);
}
/**
@@ -164,33 +212,41 @@ public final class JoinWindows extends Windows<Window> {
* @return the window maintain duration
* @deprecated since 2.1. Use {@link JoinWindows#gracePeriodMs()} instead.
*/
+ @SuppressWarnings("deprecation")
@Override
@Deprecated
public long maintainMs() {
- return Math.max(super.maintainMs(), size());
+ return Math.max(maintainDurationMs, size());
}
+ @SuppressWarnings({"deprecation", "NonFinalFieldReferenceInEquals"}) // removing segments from Windows will fix this
@Override
public boolean equals(final Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- if (!super.equals(o)) return false;
final JoinWindows that = (JoinWindows) o;
return beforeMs == that.beforeMs &&
- afterMs == that.afterMs;
+ afterMs == that.afterMs &&
+ maintainDurationMs == that.maintainDurationMs &&
+ segments == that.segments &&
+ Objects.equals(grace, that.grace);
}
+ @SuppressWarnings({"deprecation", "NonFinalFieldReferencedInHashCode"}) // removing segments from Windows will fix this
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), beforeMs, afterMs);
+ return Objects.hash(beforeMs, afterMs, grace, maintainDurationMs, segments);
}
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
@Override
public String toString() {
return "JoinWindows{" +
"beforeMs=" + beforeMs +
", afterMs=" + afterMs +
- ", super=" + super.toString() +
+ ", grace=" + grace +
+ ", maintainDurationMs=" + maintainDurationMs +
+ ", segments=" + segments +
'}';
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
index c4dd91d..9efb78c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
@@ -22,6 +22,8 @@ import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
import java.time.Duration;
import java.util.Objects;
+import static org.apache.kafka.streams.kstream.internals.WindowingDefaults.DEFAULT_RETENTION_MS;
+
/**
* A session based window specification used for aggregating events into sessions.
@@ -91,8 +93,7 @@ public final class SessionWindows {
if (inactivityGapMs <= 0) {
throw new IllegalArgumentException("Gap time (inactivityGapMs) cannot be zero or negative.");
}
- final long oneDayMs = 24 * 60 * 60_000L;
- return new SessionWindows(inactivityGapMs, oneDayMs, null);
+ return new SessionWindows(inactivityGapMs, DEFAULT_RETENTION_MS, null);
}
/**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
index 8b62a43..6a58c2c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
@@ -20,10 +20,13 @@ import org.apache.kafka.streams.kstream.internals.TimeWindow;
import org.apache.kafka.streams.processor.TimestampExtractor;
import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
+import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
+import static org.apache.kafka.streams.kstream.internals.WindowingDefaults.DEFAULT_RETENTION_MS;
+
/**
* The fixed-size time-based window specifications used for aggregations.
* <p>
@@ -52,6 +55,8 @@ import java.util.Objects;
*/
public final class TimeWindows extends Windows<TimeWindow> {
+ private final long maintainDurationMs;
+
/** The size of the windows in milliseconds. */
public final long sizeMs;
@@ -60,10 +65,28 @@ public final class TimeWindows extends Windows<TimeWindow> {
* the previous one.
*/
public final long advanceMs;
+ private final Duration grace;
+
+ private TimeWindows(final long sizeMs, final long advanceMs, final Duration grace, final long maintainDurationMs) {
+ this.sizeMs = sizeMs;
+ this.advanceMs = advanceMs;
+ this.grace = grace;
+ this.maintainDurationMs = maintainDurationMs;
+ }
- private TimeWindows(final long sizeMs, final long advanceMs) {
+ /** Private constructor for preserving segments. Can be removed along with Windows.segments. **/
+ @SuppressWarnings("DeprecatedIsStillUsed")
+ @Deprecated
+ private TimeWindows(final long sizeMs,
+ final long advanceMs,
+ final Duration grace,
+ final long maintainDurationMs,
+ final int segments) {
+ super(segments);
this.sizeMs = sizeMs;
this.advanceMs = advanceMs;
+ this.grace = grace;
+ this.maintainDurationMs = maintainDurationMs;
}
/**
@@ -82,7 +105,8 @@ public final class TimeWindows extends Windows<TimeWindow> {
if (sizeMs <= 0) {
throw new IllegalArgumentException("Window size (sizeMs) must be larger than zero.");
}
- return new TimeWindows(sizeMs, sizeMs);
+ // This is a static factory method, so we initialize grace and retention to the defaults.
+ return new TimeWindows(sizeMs, sizeMs, null, DEFAULT_RETENTION_MS);
}
/**
@@ -97,11 +121,12 @@ public final class TimeWindows extends Windows<TimeWindow> {
* @return a new window definition with default maintain duration of 1 day
* @throws IllegalArgumentException if the advance interval is negative, zero, or larger-or-equal the window size
*/
+ @SuppressWarnings("deprecation") // will be fixed when we remove segments from Windows
public TimeWindows advanceBy(final long advanceMs) {
if (advanceMs <= 0 || advanceMs > sizeMs) {
throw new IllegalArgumentException(String.format("AdvanceMs must lie within interval (0, %d].", sizeMs));
}
- return new TimeWindows(sizeMs, advanceMs);
+ return new TimeWindows(sizeMs, advanceMs, grace, maintainDurationMs, segments);
}
@Override
@@ -121,10 +146,30 @@ public final class TimeWindows extends Windows<TimeWindow> {
return sizeMs;
}
- @Override
+ /**
+ * Reject late events that arrive more than {@code millisAfterWindowEnd}
+ * after the end of its window.
+ *
+ * Lateness is defined as (stream_time - record_timestamp).
+ *
+ * @param millisAfterWindowEnd The grace period to admit late-arriving events to a window.
+ * @return this updated builder
+ */
+ @SuppressWarnings("deprecation") // will be fixed when we remove segments from Windows
public TimeWindows grace(final long millisAfterWindowEnd) {
- super.grace(millisAfterWindowEnd);
- return this;
+ if (millisAfterWindowEnd < 0) {
+ throw new IllegalArgumentException("Grace period must not be negative.");
+ }
+ return new TimeWindows(sizeMs, advanceMs, Duration.ofMillis(millisAfterWindowEnd), maintainDurationMs, segments);
+ }
+
+ @SuppressWarnings("deprecation") // continuing to support Windows#maintainMs/segmentInterval in fallback mode
+ @Override
+ public long gracePeriodMs() {
+ // NOTE: in the future, when we remove maintainMs,
+ // we should default the grace period to 24h to maintain the default behavior,
+ // or we can default to (24h - size) if you want to be super accurate.
+ return grace != null ? grace.toMillis() : maintainMs() - size();
}
/**
@@ -135,14 +180,14 @@ public final class TimeWindows extends Windows<TimeWindow> {
* @deprecated since 2.1. Use {@link Materialized#retention} or directly configure the retention in a store supplier
* and use {@link Materialized#as(WindowBytesStoreSupplier)}.
*/
+ @SuppressWarnings("deprecation")
@Override
@Deprecated
public TimeWindows until(final long durationMs) throws IllegalArgumentException {
if (durationMs < sizeMs) {
throw new IllegalArgumentException("Window retention time (durationMs) cannot be smaller than the window size.");
}
- super.until(durationMs);
- return this;
+ return new TimeWindows(sizeMs, advanceMs, grace, durationMs, segments);
}
/**
@@ -153,33 +198,41 @@ public final class TimeWindows extends Windows<TimeWindow> {
* @return the window maintain duration
* @deprecated since 2.1. Use {@link Materialized#retention} instead.
*/
+ @SuppressWarnings({"DeprecatedIsStillUsed", "deprecation"})
@Override
@Deprecated
public long maintainMs() {
- return Math.max(super.maintainMs(), sizeMs);
+ return Math.max(maintainDurationMs, sizeMs);
}
+ @SuppressWarnings({"deprecation", "NonFinalFieldReferenceInEquals"}) // removing segments from Windows will fix this
@Override
public boolean equals(final Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- if (!super.equals(o)) return false;
final TimeWindows that = (TimeWindows) o;
- return sizeMs == that.sizeMs &&
- advanceMs == that.advanceMs;
+ return maintainDurationMs == that.maintainDurationMs &&
+ segments == that.segments &&
+ sizeMs == that.sizeMs &&
+ advanceMs == that.advanceMs &&
+ Objects.equals(grace, that.grace);
}
+ @SuppressWarnings({"deprecation", "NonFinalFieldReferencedInHashCode"}) // removing segments from Windows will fix this
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), sizeMs, advanceMs);
+ return Objects.hash(maintainDurationMs, segments, sizeMs, advanceMs, grace);
}
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
@Override
public String toString() {
return "TimeWindows{" +
- "sizeMs=" + sizeMs +
+ "maintainDurationMs=" + maintainDurationMs +
+ ", sizeMs=" + sizeMs +
", advanceMs=" + advanceMs +
- ", super=" + super.toString() +
+ ", grace=" + grace +
+ ", segments=" + segments +
'}';
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
index cf4482f..73aa9b1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
@@ -100,6 +100,7 @@ public final class UnlimitedWindows extends Windows<UnlimitedWindow> {
* @throws IllegalArgumentException on every invocation.
* @deprecated since 2.1.
*/
+ @SuppressWarnings("deprecation")
@Override
@Deprecated
public UnlimitedWindows until(final long durationMs) {
@@ -113,42 +114,39 @@ public final class UnlimitedWindows extends Windows<UnlimitedWindow> {
* @return the window retention time that is {@link Long#MAX_VALUE}
* @deprecated since 2.1. Use {@link Materialized#retention} instead.
*/
+ @SuppressWarnings("deprecation")
@Override
@Deprecated
public long maintainMs() {
return Long.MAX_VALUE;
}
- /**
- * Throws an {@link IllegalArgumentException} because the window never ends and the
- * grace period is therefore meaningless.
- *
- * @throws IllegalArgumentException on every invocation
- */
@Override
- public UnlimitedWindows grace(final long millisAfterWindowEnd) {
- throw new IllegalArgumentException("Grace period cannot be set for UnlimitedWindows.");
+ public long gracePeriodMs() {
+ return 0L;
}
+ @SuppressWarnings({"deprecation", "NonFinalFieldReferenceInEquals"}) // removing segments from Windows will fix this
@Override
public boolean equals(final Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- if (!super.equals(o)) return false;
final UnlimitedWindows that = (UnlimitedWindows) o;
- return startMs == that.startMs;
+ return startMs == that.startMs && segments == that.segments;
}
+ @SuppressWarnings({"deprecation", "NonFinalFieldReferencedInHashCode"}) // removing segments from Windows will fix this
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), startMs);
+ return Objects.hash(startMs, segments);
}
+ @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
@Override
public String toString() {
return "UnlimitedWindows{" +
"startMs=" + startMs +
- ", super=" + super.toString() +
+ ", segments=" + segments +
'}';
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
index 8c9ba55..b6b2178 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
@@ -19,9 +19,9 @@ package org.apache.kafka.streams.kstream;
import org.apache.kafka.streams.processor.TimestampExtractor;
import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
-import java.time.Duration;
import java.util.Map;
-import java.util.Objects;
+
+import static org.apache.kafka.streams.kstream.internals.WindowingDefaults.DEFAULT_RETENTION_MS;
/**
* The window specification for fixed size windows that is used to define window boundaries and grace period.
@@ -40,44 +40,14 @@ import java.util.Objects;
*/
public abstract class Windows<W extends Window> {
- private long maintainDurationMs = 24 * 60 * 60 * 1000L; // default: one day
+ private long maintainDurationMs = DEFAULT_RETENTION_MS;
@Deprecated public int segments = 3;
- private Duration grace;
-
protected Windows() {}
- /**
- * Reject late events that arrive more than {@code millisAfterWindowEnd}
- * after the end of its window.
- *
- * Lateness is defined as (stream_time - record_timestamp).
- *
- * @param millisAfterWindowEnd The grace period to admit late-arriving events to a window.
- * @return this updated builder
- */
- public Windows<W> grace(final long millisAfterWindowEnd) {
- if (millisAfterWindowEnd < 0) {
- throw new IllegalArgumentException("Grace period must not be negative.");
- }
-
- grace = Duration.ofMillis(millisAfterWindowEnd);
-
- return this;
- }
-
- /**
- * Return the window grace period (the time to admit
- * late-arriving events after the end of the window.)
- *
- * Lateness is defined as (stream_time - record_timestamp).
- */
- @SuppressWarnings("deprecation") // continuing to support Windows#maintainMs/segmentInterval in fallback mode
- public long gracePeriodMs() {
- // NOTE: in the future, when we remove maintainMs,
- // we should default the grace period to 24h to maintain the default behavior,
- // or we can default to (24h - size) if you want to be super accurate.
- return grace != null ? grace.toMillis() : maintainMs() - size();
+ @SuppressWarnings("deprecation") // remove this constructor when we remove segments.
+ Windows(final int segments) {
+ this.segments = segments;
}
/**
@@ -106,6 +76,7 @@ public abstract class Windows<W extends Window> {
* @return the window maintain duration
* @deprecated since 2.1. Use {@link Materialized#retention} instead.
*/
+ @SuppressWarnings("DeprecatedIsStillUsed")
@Deprecated
public long maintainMs() {
return maintainDurationMs;
@@ -161,34 +132,10 @@ public abstract class Windows<W extends Window> {
public abstract long size();
/**
- * Warning: It may be unsafe to use objects of this class in set- or map-like collections,
- * since the equals and hashCode methods depend on mutable fields.
- */
- @Override
- public boolean equals(final Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- final Windows<?> windows = (Windows<?>) o;
- return maintainMs() == windows.maintainMs() &&
- segments == windows.segments &&
- Objects.equals(gracePeriodMs(), windows.gracePeriodMs());
- }
-
- /**
- * Warning: It may be unsafe to use objects of this class in set- or map-like collections,
- * since the equals and hashCode methods depend on mutable fields.
+ * Return the window grace period (the time to admit
+ * late-arriving events after the end of the window.)
+ *
+ * Lateness is defined as (stream_time - record_timestamp).
*/
- @Override
- public int hashCode() {
- return Objects.hash(maintainMs(), segments, gracePeriodMs());
- }
-
- @Override
- public String toString() {
- return "Windows{" +
- "maintainDurationMs=" + maintainMs() +
- ", segments=" + segments +
- ", grace=" + gracePeriodMs() +
- '}';
- }
+ public abstract long gracePeriodMs();
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/WindowingDefaults.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/WindowingDefaults.java
new file mode 100644
index 0000000..50015c1
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/WindowingDefaults.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.kstream.internals;
+
+public final class WindowingDefaults {
+ private WindowingDefaults() {}
+
+ public static final long DEFAULT_RETENTION_MS = 24 * 60 * 60 * 1000L; // one day
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/EqualityCheck.java b/streams/src/test/java/org/apache/kafka/streams/EqualityCheck.java
new file mode 100644
index 0000000..4ca1dde
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/EqualityCheck.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams;
+
+public final class EqualityCheck {
+ private EqualityCheck() {}
+
+ // Inspired by EqualsTester from Guava
+ public static <T> void verifyEquality(final T o1, final T o2) {
+ // making sure we don't get an NPE in the test
+ if (o1 == null && o2 == null) {
+ return;
+ } else if (o1 == null) {
+ throw new AssertionError(String.format("o1 was null, but o2[%s] was not.", o2));
+ } else if (o2 == null) {
+ throw new AssertionError(String.format("o1[%s] was not null, but o2 was.", o1));
+ }
+ verifyGeneralEqualityProperties(o1, o2);
+
+
+ // the two objects should equal each other
+ if (!o1.equals(o2)) {
+ throw new AssertionError(String.format("o1[%s] was not equal to o2[%s].", o1, o2));
+ }
+
+ if (!o2.equals(o1)) {
+ throw new AssertionError(String.format("o2[%s] was not equal to o1[%s].", o2, o1));
+ }
+
+ verifyHashCodeConsistency(o1, o2);
+
+ // since these objects are equal, their hashcode MUST be the same
+ if (o1.hashCode() != o2.hashCode()) {
+ throw new AssertionError(String.format("o1[%s].hash[%d] was not equal to o2[%s].hash[%d].", o1, o1.hashCode(), o2, o2.hashCode()));
+ }
+ }
+
+ public static <T> void verifyInEquality(final T o1, final T o2) {
+ // making sure we don't get an NPE in the test
+ if (o1 == null && o2 == null) {
+ throw new AssertionError("Both o1 and o2 were null.");
+ } else if (o1 == null) {
+ return;
+ } else if (o2 == null) {
+ return;
+ }
+
+ verifyGeneralEqualityProperties(o1, o2);
+
+ // these two objects should NOT equal each other
+ if (o1.equals(o2)) {
+ throw new AssertionError(String.format("o1[%s] was equal to o2[%s].", o1, o2));
+ }
+
+ if (o2.equals(o1)) {
+ throw new AssertionError(String.format("o2[%s] was equal to o1[%s].", o2, o1));
+ }
+ verifyHashCodeConsistency(o1, o2);
+
+
+ // since these objects are NOT equal, their hashcode SHOULD PROBABLY not be the same
+ if (o1.hashCode() == o2.hashCode()) {
+ throw new AssertionError(
+ String.format(
+ "o1[%s].hash[%d] was equal to o2[%s].hash[%d], even though !o1.equals(o2). " +
+ "This is NOT A BUG, but it is undesirable for hash collection performance.",
+ o1,
+ o1.hashCode(),
+ o2,
+ o2.hashCode()
+ )
+ );
+ }
+ }
+
+
+ @SuppressWarnings({"EqualsWithItself", "ConstantConditions", "ObjectEqualsNull"})
+ private static <T> void verifyGeneralEqualityProperties(final T o1, final T o2) {
+ // objects should equal themselves
+ if (!o1.equals(o1)) {
+ throw new AssertionError(String.format("o1[%s] was not equal to itself.", o1));
+ }
+
+ if (!o2.equals(o2)) {
+ throw new AssertionError(String.format("o2[%s] was not equal to itself.", o2));
+ }
+
+ // non-null objects should not equal null
+ if (o1.equals(null)) {
+ throw new AssertionError(String.format("o1[%s] was equal to null.", o1));
+ }
+
+ if (o2.equals(null)) {
+ throw new AssertionError(String.format("o2[%s] was equal to null.", o2));
+ }
+
+ // objects should not equal some random object
+ if (o1.equals(new Object())) {
+ throw new AssertionError(String.format("o1[%s] was equal to an anonymous Object.", o1));
+ }
+
+ if (o2.equals(new Object())) {
+ throw new AssertionError(String.format("o2[%s] was equal to an anonymous Object.", o2));
+ }
+ }
+
+
+ private static <T> void verifyHashCodeConsistency(final T o1, final T o2) {
+ {
+ final int first = o1.hashCode();
+ final int second = o1.hashCode();
+ if (first != second) {
+ throw new AssertionError(
+ String.format(
+ "o1[%s]'s hashcode was not consistent: [%d]!=[%d].",
+ o1,
+ first,
+ second
+ )
+ );
+ }
+ }
+
+ {
+ final int first = o2.hashCode();
+ final int second = o2.hashCode();
+ if (first != second) {
+ throw new AssertionError(
+ String.format(
+ "o2[%s]'s hashcode was not consistent: [%d]!=[%d].",
+ o2,
+ first,
+ second
+ )
+ );
+ }
+ }
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
index c16950e..ce6c352 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamAggregationIntegrationTest.java
@@ -49,10 +49,12 @@ import org.apache.kafka.streams.kstream.SessionWindows;
import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Transformer;
+import org.apache.kafka.streams.kstream.UnlimitedWindows;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.WindowedSerdes;
import org.apache.kafka.streams.kstream.internals.SessionWindow;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
+import org.apache.kafka.streams.kstream.internals.UnlimitedWindow;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -70,6 +72,7 @@ import org.junit.experimental.categories.Category;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
@@ -645,6 +648,93 @@ public class KStreamAggregationIntegrationTest {
}
+ @Test
+ public void shouldCountUnlimitedWindows() throws Exception {
+ final long startTime = mockTime.milliseconds() - TimeUnit.MILLISECONDS.convert(1, TimeUnit.HOURS) + 1;
+ final long incrementTime = Duration.ofDays(1).toMillis();
+
+ final long t1 = mockTime.milliseconds() - TimeUnit.MILLISECONDS.convert(1, TimeUnit.HOURS);
+ final List<KeyValue<String, String>> t1Messages = Arrays.asList(new KeyValue<>("bob", "start"),
+ new KeyValue<>("penny", "start"),
+ new KeyValue<>("jo", "pause"),
+ new KeyValue<>("emily", "pause"));
+
+ final Properties producerConfig = TestUtils.producerConfig(
+ CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class,
+ new Properties()
+ );
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ userSessionsStream,
+ t1Messages,
+ producerConfig,
+ t1);
+
+ final long t2 = t1 + incrementTime;
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ userSessionsStream,
+ Collections.singletonList(
+ new KeyValue<>("emily", "resume")
+ ),
+ producerConfig,
+ t2);
+ final long t3 = t2 + incrementTime;
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ userSessionsStream,
+ Arrays.asList(
+ new KeyValue<>("bob", "pause"),
+ new KeyValue<>("penny", "stop")
+ ),
+ producerConfig,
+ t3);
+
+ final long t4 = t3 + incrementTime;
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ userSessionsStream,
+ Arrays.asList(
+ new KeyValue<>("bob", "resume"), // bobs session continues
+ new KeyValue<>("jo", "resume") // jo's starts new session
+ ),
+ producerConfig,
+ t4);
+
+ final Map<Windowed<String>, KeyValue<Long, Long>> results = new HashMap<>();
+ final CountDownLatch latch = new CountDownLatch(5);
+
+ builder.stream(userSessionsStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .groupByKey(Serialized.with(Serdes.String(), Serdes.String()))
+ .windowedBy(UnlimitedWindows.of().startOn(startTime))
+ .count()
+ .toStream()
+ .transform(() -> new Transformer<Windowed<String>, Long, KeyValue<Object, Object>>() {
+ private ProcessorContext context;
+
+ @Override
+ public void init(final ProcessorContext context) {
+ this.context = context;
+ }
+
+ @Override
+ public KeyValue<Object, Object> transform(final Windowed<String> key, final Long value) {
+ results.put(key, KeyValue.pair(value, context.timestamp()));
+ latch.countDown();
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ });
+
+ startStreams();
+ assertTrue(latch.await(30, TimeUnit.SECONDS));
+ assertThat(results.get(new Windowed<>("bob", new UnlimitedWindow(startTime))), equalTo(KeyValue.pair(2L, t4)));
+ assertThat(results.get(new Windowed<>("penny", new UnlimitedWindow(startTime))), equalTo(KeyValue.pair(1L, t3)));
+ assertThat(results.get(new Windowed<>("jo", new UnlimitedWindow(startTime))), equalTo(KeyValue.pair(1L, t4)));
+ assertThat(results.get(new Windowed<>("emily", new UnlimitedWindow(startTime))), equalTo(KeyValue.pair(1L, t2)));
+ }
+
private void produceMessages(final long timestamp) throws Exception {
IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java
index ae79371..de635b4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/JoinWindowsTest.java
@@ -18,8 +18,9 @@ package org.apache.kafka.streams.kstream;
import org.junit.Test;
+import static org.apache.kafka.streams.EqualityCheck.verifyEquality;
+import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.fail;
@@ -105,92 +106,58 @@ public class JoinWindowsTest {
@Test
public void equalsAndHashcodeShouldBeValidForPositiveCases() {
- assertEquals(JoinWindows.of(3), JoinWindows.of(3));
- assertEquals(JoinWindows.of(3).hashCode(), JoinWindows.of(3).hashCode());
+ verifyEquality(JoinWindows.of(3), JoinWindows.of(3));
- assertEquals(JoinWindows.of(3).after(2), JoinWindows.of(3).after(2));
- assertEquals(JoinWindows.of(3).after(2).hashCode(), JoinWindows.of(3).after(2).hashCode());
+ verifyEquality(JoinWindows.of(3).after(2), JoinWindows.of(3).after(2));
- assertEquals(JoinWindows.of(3).before(2), JoinWindows.of(3).before(2));
- assertEquals(JoinWindows.of(3).before(2).hashCode(), JoinWindows.of(3).before(2).hashCode());
+ verifyEquality(JoinWindows.of(3).before(2), JoinWindows.of(3).before(2));
- assertEquals(JoinWindows.of(3).grace(2), JoinWindows.of(3).grace(2));
- assertEquals(JoinWindows.of(3).grace(2).hashCode(), JoinWindows.of(3).grace(2).hashCode());
+ verifyEquality(JoinWindows.of(3).grace(2), JoinWindows.of(3).grace(2));
- assertEquals(JoinWindows.of(3).until(60), JoinWindows.of(3).until(60));
- assertEquals(JoinWindows.of(3).until(60).hashCode(), JoinWindows.of(3).until(60).hashCode());
+ verifyEquality(JoinWindows.of(3).until(60), JoinWindows.of(3).until(60));
- assertEquals(
+ verifyEquality(
JoinWindows.of(3).before(1).after(2).grace(3).until(60),
JoinWindows.of(3).before(1).after(2).grace(3).until(60)
);
- assertEquals(
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode(),
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode()
- );
// JoinWindows is a little weird in that before and after set the same fields as of.
- assertEquals(
+ verifyEquality(
JoinWindows.of(9).before(1).after(2).grace(3).until(60),
JoinWindows.of(3).before(1).after(2).grace(3).until(60)
);
- assertEquals(
- JoinWindows.of(9).before(1).after(2).grace(3).until(60).hashCode(),
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode()
- );
}
@Test
public void equalsAndHashcodeShouldBeValidForNegativeCases() {
- assertNotEquals(JoinWindows.of(9), JoinWindows.of(3));
- assertNotEquals(JoinWindows.of(9).hashCode(), JoinWindows.of(3).hashCode());
+ verifyInEquality(JoinWindows.of(9), JoinWindows.of(3));
- assertNotEquals(JoinWindows.of(3).after(9), JoinWindows.of(3).after(2));
- assertNotEquals(JoinWindows.of(3).after(9).hashCode(), JoinWindows.of(3).after(2).hashCode());
+ verifyInEquality(JoinWindows.of(3).after(9), JoinWindows.of(3).after(2));
- assertNotEquals(JoinWindows.of(3).before(9), JoinWindows.of(3).before(2));
- assertNotEquals(JoinWindows.of(3).before(9).hashCode(), JoinWindows.of(3).before(2).hashCode());
+ verifyInEquality(JoinWindows.of(3).before(9), JoinWindows.of(3).before(2));
- assertNotEquals(JoinWindows.of(3).grace(9), JoinWindows.of(3).grace(2));
- assertNotEquals(JoinWindows.of(3).grace(9).hashCode(), JoinWindows.of(3).grace(2).hashCode());
+ verifyInEquality(JoinWindows.of(3).grace(9), JoinWindows.of(3).grace(2));
- assertNotEquals(JoinWindows.of(3).until(90), JoinWindows.of(3).until(60));
- assertNotEquals(JoinWindows.of(3).until(90).hashCode(), JoinWindows.of(3).until(60).hashCode());
+ verifyInEquality(JoinWindows.of(3).until(90), JoinWindows.of(3).until(60));
- assertNotEquals(
+ verifyInEquality(
JoinWindows.of(3).before(9).after(2).grace(3).until(60),
JoinWindows.of(3).before(1).after(2).grace(3).until(60)
);
- assertNotEquals(
- JoinWindows.of(3).before(9).after(2).grace(3).until(60).hashCode(),
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode()
- );
- assertNotEquals(
+ verifyInEquality(
JoinWindows.of(3).before(1).after(9).grace(3).until(60),
JoinWindows.of(3).before(1).after(2).grace(3).until(60)
);
- assertNotEquals(
- JoinWindows.of(3).before(1).after(9).grace(3).until(60).hashCode(),
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode()
- );
- assertNotEquals(
+ verifyInEquality(
JoinWindows.of(3).before(1).after(2).grace(9).until(60),
JoinWindows.of(3).before(1).after(2).grace(3).until(60)
);
- assertNotEquals(
- JoinWindows.of(3).before(1).after(2).grace(9).until(60).hashCode(),
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode()
- );
- assertNotEquals(
+ verifyInEquality(
JoinWindows.of(3).before(1).after(2).grace(3).until(90),
JoinWindows.of(3).before(1).after(2).grace(3).until(60)
);
- assertNotEquals(
- JoinWindows.of(3).before(1).after(2).grace(3).until(90).hashCode(),
- JoinWindows.of(3).before(1).after(2).grace(3).until(60).hashCode()
- );
}
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java
index 2acd5d2..9f99be4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java
@@ -18,8 +18,9 @@ package org.apache.kafka.streams.kstream;
import org.junit.Test;
+import static org.apache.kafka.streams.EqualityCheck.verifyEquality;
+import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.fail;
public class SessionWindowsTest {
@@ -81,38 +82,27 @@ public class SessionWindowsTest {
@Test
public void equalsAndHashcodeShouldBeValidForPositiveCases() {
- assertEquals(SessionWindows.with(1), SessionWindows.with(1));
- assertEquals(SessionWindows.with(1).hashCode(), SessionWindows.with(1).hashCode());
+ verifyEquality(SessionWindows.with(1), SessionWindows.with(1));
- assertEquals(SessionWindows.with(1).grace(6), SessionWindows.with(1).grace(6));
- assertEquals(SessionWindows.with(1).grace(6).hashCode(), SessionWindows.with(1).grace(6).hashCode());
+ verifyEquality(SessionWindows.with(1).grace(6), SessionWindows.with(1).grace(6));
- assertEquals(SessionWindows.with(1).until(7), SessionWindows.with(1).until(7));
- assertEquals(SessionWindows.with(1).until(7).hashCode(), SessionWindows.with(1).until(7).hashCode());
+ verifyEquality(SessionWindows.with(1).until(7), SessionWindows.with(1).until(7));
- assertEquals(SessionWindows.with(1).grace(6).until(7), SessionWindows.with(1).grace(6).until(7));
- assertEquals(SessionWindows.with(1).grace(6).until(7).hashCode(), SessionWindows.with(1).grace(6).until(7).hashCode());
+ verifyEquality(SessionWindows.with(1).grace(6).until(7), SessionWindows.with(1).grace(6).until(7));
}
@Test
public void equalsAndHashcodeShouldBeValidForNegativeCases() {
- assertNotEquals(SessionWindows.with(9), SessionWindows.with(1));
- assertNotEquals(SessionWindows.with(9).hashCode(), SessionWindows.with(1).hashCode());
+ verifyInEquality(SessionWindows.with(9), SessionWindows.with(1));
- assertNotEquals(SessionWindows.with(1).grace(9), SessionWindows.with(1).grace(6));
- assertNotEquals(SessionWindows.with(1).grace(9).hashCode(), SessionWindows.with(1).grace(6).hashCode());
+ verifyInEquality(SessionWindows.with(1).grace(9), SessionWindows.with(1).grace(6));
- assertNotEquals(SessionWindows.with(1).until(9), SessionWindows.with(1).until(7));
- assertNotEquals(SessionWindows.with(1).until(9).hashCode(), SessionWindows.with(1).until(7).hashCode());
+ verifyInEquality(SessionWindows.with(1).until(9), SessionWindows.with(1).until(7));
+ verifyInEquality(SessionWindows.with(2).grace(6).until(7), SessionWindows.with(1).grace(6).until(7));
- assertNotEquals(SessionWindows.with(2).grace(6).until(7), SessionWindows.with(1).grace(6).until(7));
- assertNotEquals(SessionWindows.with(2).grace(6).until(7).hashCode(), SessionWindows.with(1).grace(6).until(7).hashCode());
+ verifyInEquality(SessionWindows.with(1).grace(0).until(7), SessionWindows.with(1).grace(6).until(7));
- assertNotEquals(SessionWindows.with(1).grace(0).until(7), SessionWindows.with(1).grace(6).until(7));
- assertNotEquals(SessionWindows.with(1).grace(0).until(7).hashCode(), SessionWindows.with(1).grace(6).until(7).hashCode());
-
- assertNotEquals(SessionWindows.with(1).grace(6).until(70), SessionWindows.with(1).grace(6).until(7));
- assertNotEquals(SessionWindows.with(1).grace(6).until(70).hashCode(), SessionWindows.with(1).grace(6).until(7).hashCode());
+ verifyInEquality(SessionWindows.with(1).grace(6).until(70), SessionWindows.with(1).grace(6).until(7));
}
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java
index d426b28..9010bb2 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/TimeWindowsTest.java
@@ -21,6 +21,8 @@ import org.junit.Test;
import java.util.Map;
+import static org.apache.kafka.streams.EqualityCheck.verifyEquality;
+import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.fail;
@@ -150,77 +152,49 @@ public class TimeWindowsTest {
@Test
public void equalsAndHashcodeShouldBeValidForPositiveCases() {
- assertEquals(TimeWindows.of(3), TimeWindows.of(3));
- assertEquals(TimeWindows.of(3).hashCode(), TimeWindows.of(3).hashCode());
+ verifyEquality(TimeWindows.of(3), TimeWindows.of(3));
- assertEquals(TimeWindows.of(3).advanceBy(1), TimeWindows.of(3).advanceBy(1));
- assertEquals(TimeWindows.of(3).advanceBy(1).hashCode(), TimeWindows.of(3).advanceBy(1).hashCode());
+ verifyEquality(TimeWindows.of(3).advanceBy(1), TimeWindows.of(3).advanceBy(1));
- assertEquals(TimeWindows.of(3).grace(1), TimeWindows.of(3).grace(1));
- assertEquals(TimeWindows.of(3).grace(1).hashCode(), TimeWindows.of(3).grace(1).hashCode());
+ verifyEquality(TimeWindows.of(3).grace(1), TimeWindows.of(3).grace(1));
- assertEquals(TimeWindows.of(3).until(4), TimeWindows.of(3).until(4));
- assertEquals(TimeWindows.of(3).until(4).hashCode(), TimeWindows.of(3).until(4).hashCode());
+ verifyEquality(TimeWindows.of(3).until(4), TimeWindows.of(3).until(4));
- assertEquals(
+ verifyEquality(
TimeWindows.of(3).advanceBy(1).grace(1).until(4),
TimeWindows.of(3).advanceBy(1).grace(1).until(4)
);
- assertEquals(
- TimeWindows.of(3).advanceBy(1).grace(1).until(4).hashCode(),
- TimeWindows.of(3).advanceBy(1).grace(1).until(4).hashCode()
- );
}
@Test
public void equalsAndHashcodeShouldBeValidForNegativeCases() {
- assertNotEquals(TimeWindows.of(9), TimeWindows.of(3));
- assertNotEquals(TimeWindows.of(9).hashCode(), TimeWindows.of(3).hashCode());
+ verifyInEquality(TimeWindows.of(9), TimeWindows.of(3));
- assertNotEquals(TimeWindows.of(3).advanceBy(2), TimeWindows.of(3).advanceBy(1));
- assertNotEquals(TimeWindows.of(3).advanceBy(2).hashCode(), TimeWindows.of(3).advanceBy(1).hashCode());
+ verifyInEquality(TimeWindows.of(3).advanceBy(2), TimeWindows.of(3).advanceBy(1));
- assertNotEquals(TimeWindows.of(3).grace(2), TimeWindows.of(3).grace(1));
- assertNotEquals(TimeWindows.of(3).grace(2).hashCode(), TimeWindows.of(3).grace(1).hashCode());
+ verifyInEquality(TimeWindows.of(3).grace(2), TimeWindows.of(3).grace(1));
- assertNotEquals(TimeWindows.of(3).until(9), TimeWindows.of(3).until(4));
- assertNotEquals(TimeWindows.of(3).until(9).hashCode(), TimeWindows.of(3).until(4).hashCode());
+ verifyInEquality(TimeWindows.of(3).until(9), TimeWindows.of(3).until(4));
- assertNotEquals(
+ verifyInEquality(
TimeWindows.of(4).advanceBy(2).grace(2).until(4),
TimeWindows.of(3).advanceBy(2).grace(2).until(4)
);
- assertNotEquals(
- TimeWindows.of(4).advanceBy(2).grace(2).until(4).hashCode(),
- TimeWindows.of(3).advanceBy(2).grace(2).until(4).hashCode()
- );
- assertNotEquals(
+ verifyInEquality(
TimeWindows.of(3).advanceBy(1).grace(2).until(4),
TimeWindows.of(3).advanceBy(2).grace(2).until(4)
);
- assertNotEquals(
- TimeWindows.of(3).advanceBy(1).grace(2).until(4).hashCode(),
- TimeWindows.of(3).advanceBy(2).grace(2).until(4).hashCode()
- );
assertNotEquals(
TimeWindows.of(3).advanceBy(2).grace(1).until(4),
TimeWindows.of(3).advanceBy(2).grace(2).until(4)
);
- assertNotEquals(
- TimeWindows.of(3).advanceBy(2).grace(1).until(4).hashCode(),
- TimeWindows.of(3).advanceBy(2).grace(2).until(4).hashCode()
- );
assertNotEquals(
TimeWindows.of(3).advanceBy(2).grace(2).until(9),
TimeWindows.of(3).advanceBy(2).grace(2).until(4)
);
- assertNotEquals(
- TimeWindows.of(3).advanceBy(2).grace(2).until(9).hashCode(),
- TimeWindows.of(3).advanceBy(2).grace(2).until(4).hashCode()
- );
}
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/UnlimitedWindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/UnlimitedWindowsTest.java
index 5d139c8..a140654 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/UnlimitedWindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/UnlimitedWindowsTest.java
@@ -21,8 +21,9 @@ import org.junit.Test;
import java.util.Map;
+import static org.apache.kafka.streams.EqualityCheck.verifyEquality;
+import static org.apache.kafka.streams.EqualityCheck.verifyInEquality;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -52,16 +53,6 @@ public class UnlimitedWindowsTest {
}
@Test
- public void gracePeriodShouldNotBeSettable() {
- try {
- UnlimitedWindows.of().grace(0L);
- fail("should not be able to set grace period");
- } catch (final IllegalArgumentException e) {
- // expected
- }
- }
-
- @Test
public void shouldIncludeRecordsThatHappenedOnWindowStart() {
final UnlimitedWindows w = UnlimitedWindows.of().startOn(anyStartTime);
final Map<Long, UnlimitedWindow> matchedWindows = w.windowsFor(w.startMs);
@@ -88,18 +79,15 @@ public class UnlimitedWindowsTest {
@Test
public void equalsAndHashcodeShouldBeValidForPositiveCases() {
- assertEquals(UnlimitedWindows.of(), UnlimitedWindows.of());
- assertEquals(UnlimitedWindows.of().hashCode(), UnlimitedWindows.of().hashCode());
+ verifyEquality(UnlimitedWindows.of(), UnlimitedWindows.of());
- assertEquals(UnlimitedWindows.of().startOn(1), UnlimitedWindows.of().startOn(1));
- assertEquals(UnlimitedWindows.of().startOn(1).hashCode(), UnlimitedWindows.of().startOn(1).hashCode());
+ verifyEquality(UnlimitedWindows.of().startOn(1), UnlimitedWindows.of().startOn(1));
}
@Test
public void equalsAndHashcodeShouldBeValidForNegativeCases() {
- assertNotEquals(UnlimitedWindows.of().startOn(9), UnlimitedWindows.of().startOn(1));
- assertNotEquals(UnlimitedWindows.of().startOn(9).hashCode(), UnlimitedWindows.of().startOn(1).hashCode());
+ verifyInEquality(UnlimitedWindows.of().startOn(9), UnlimitedWindows.of().startOn(1));
}
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/WindowsTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/WindowsTest.java
index 12ff166..8709031 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/WindowsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/WindowsTest.java
@@ -21,12 +21,10 @@ import org.junit.Test;
import java.util.Map;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
public class WindowsTest {
private class TestWindows extends Windows {
-
@Override
public Map windowsFor(final long timestamp) {
return null;
@@ -36,6 +34,10 @@ public class WindowsTest {
public long size() {
return 0;
}
+
+ public long gracePeriodMs() {
+ return 0L;
+ }
}
@SuppressWarnings("deprecation") // specifically testing deprecated APIs
@@ -58,19 +60,6 @@ public class WindowsTest {
assertEquals(anyNotNegativeRetentionTime, new TestWindows().until(anyNotNegativeRetentionTime).maintainMs());
}
-
- @Test
- public void gracePeriodShouldEnforceBoundaries() {
- new TestWindows().grace(0L);
-
- try {
- new TestWindows().grace(-1L);
- fail("should not accept negatives");
- } catch (final IllegalArgumentException e) {
- //expected
- }
- }
-
@SuppressWarnings("deprecation") // specifically testing deprecated APIs
@Test(expected = IllegalArgumentException.class)
public void numberOfSegmentsMustBeAtLeastTwo() {