MINOR: add equals to SessionWindows

Author: Damian Guy <damian.guy@gmail.com>

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Matthias J. Sax<matthias@confluent.io>, Bill Bejeck <bill@confluent.io>

Closes #4074 from dguy/minor-session-window-equals

streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java

@@ -18,6 +18,8 @@ package org.apache.kafka.streams.kstream;
 
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
+import java.util.Objects;
+
 /**
  * A session based window specification used for aggregating events into sessions.
  * <p>
@@ -121,4 +123,18 @@ public final class SessionWindows {
     public long maintainMs() {
         return Math.max(maintainDurationMs, gapMs);
     }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        final SessionWindows that = (SessionWindows) o;
+        return gapMs == that.gapMs &&
+                maintainDurationMs == that.maintainDurationMs;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(gapMs, maintainDurationMs);
+    }
 }

streams/src/test/java/org/apache/kafka/streams/kstream/SessionWindowsTest.java

@@ -18,6 +18,9 @@ package org.apache.kafka.streams.kstream;
 
 import org.junit.Test;
 
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsNot.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
@@ -62,4 +65,18 @@ public class SessionWindowsTest {
         }
     }
 
+    @Test
+    public void shouldBeEqualWhenGapAndMaintainMsAreTheSame() {
+        assertThat(SessionWindows.with(5), equalTo(SessionWindows.with(5)));
+    }
+
+    @Test
+    public void shouldNotBeEqualWhenMaintainMsDifferent() {
+        assertThat(SessionWindows.with(5), not(equalTo(SessionWindows.with(5).until(10))));
+    }
+
+    @Test
+    public void shouldNotBeEqualWhenGapIsDifferent() {
+        assertThat(SessionWindows.with(5), not(equalTo(SessionWindows.with(10))));
+    }
 }