From aea3d72aa5f6854a845e76eb4359ca456ae7209f Mon Sep 17 00:00:00 2001
From: coscale_kdegroot <koen.degroote@coscale.com>
Date: Tue, 19 Sep 2017 17:34:32 +0200
Subject: [PATCH] Added equals() method to Stamped, because a class that
 implements Comparable should always have this. findBugs complained about
 PunctuationSchedula, a subclass of Stamped, nothing having equals, so also
 implemented that.

Since the compareTo and equals methods don't take the value field into account, neither should hashCode.
---
 .../internals/PunctuationSchedule.java        | 36 +++++++++++++++++++
 .../streams/processor/internals/Stamped.java  | 27 ++++++++++++++
 2 files changed, 63 insertions(+)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/PunctuationSchedule.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/PunctuationSchedule.java
index cf50005fcd1eb..af4cde38a299b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/PunctuationSchedule.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/PunctuationSchedule.java
@@ -85,4 +85,40 @@ synchronized public void cancel() {
             schedule.markCancelled();
         }
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+
+        PunctuationSchedule that = (PunctuationSchedule) o;
+
+        if (interval != that.interval) {
+            return false;
+        }
+        if (isCancelled != that.isCancelled) {
+            return false;
+        }
+        if (punctuator != null ? !punctuator.equals(that.punctuator) : that.punctuator != null) {
+            return false;
+        }
+        return cancellable != null ? cancellable.equals(that.cancellable) : that.cancellable == null;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = super.hashCode();
+        result = 31 * result + (int) (interval ^ (interval >>> 32));
+        result = 31 * result + (punctuator != null ? punctuator.hashCode() : 0);
+        result = 31 * result + (isCancelled ? 1 : 0);
+        result = 31 * result + (cancellable != null ? cancellable.hashCode() : 0);
+        return result;
+    }
 }
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Stamped.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Stamped.java
index 4965238e8568f..311d6376810e5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Stamped.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Stamped.java
@@ -34,4 +34,31 @@ public int compareTo(Object other) {
         else if (timestamp > otherTimestamp) return 1;
         return 0;
     }
+
+    /*
+     * public classes implementing Comparable should always implement both compareTo() and equals().
+     * This is because an end-user can at some point add objects of that class to java.util.SortedSet.
+     * If the compareTo() and equals() implementations are not consistent, that would violate the contract of java.util.Set, which is defined in terms of equals().
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        Stamped<?> stamped = (Stamped<?>) o;
+
+        if (timestamp != stamped.timestamp) {
+            return false;
+        }
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        return (int) (timestamp ^ (timestamp >>> 32));
+    }
 }