[FLINK-8561] [cep] Fix SharedBuffer.removeEdges to use .equals().

This closes apache#5414.

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/SharedBuffer.java

@@ -175,7 +175,7 @@ public boolean isEmpty() {
 	 * @return {@code true} if pruning happened
 	 */
 	public boolean prune(long pruningTimestamp) {
-		final List<SharedBufferEntry<K, V>> prunedEntries = new ArrayList<>();
+		final Set<SharedBufferEntry<K, V>> prunedEntries = new HashSet<>();
 
 		final Iterator<Map.Entry<K, SharedBufferPage<K, V>>> it = pages.entrySet().iterator();
 		while (it.hasNext()) {
@@ -428,7 +428,7 @@ public SharedBufferEntry<K, V> get(final ValueTimeWrapper<V> valueTime) {
 		 * @param pruningTimestamp Timestamp for the pruning
 		 * @param prunedEntries a {@link Set} to put the removed {@link SharedBufferEntry SharedBufferEntries}.
 		 */
-		private void prune(final long pruningTimestamp, final List<SharedBufferEntry<K, V>> prunedEntries) {
+		private void prune(final long pruningTimestamp, final Set<SharedBufferEntry<K, V>> prunedEntries) {
 			Iterator<Map.Entry<ValueTimeWrapper<V>, SharedBufferEntry<K, V>>> it = entries.entrySet().iterator();
 			while (it.hasNext()) {
 				SharedBufferEntry<K, V> entry = it.next().getValue();
@@ -442,7 +442,7 @@ private void prune(final long pruningTimestamp, final List<SharedBufferEntry<K,
 		/**
 		 * Remove edges with the specified targets for the entries.
 		 */
-		private void removeEdges(final List<SharedBufferEntry<K, V>> prunedEntries) {
+		private void removeEdges(final Set<SharedBufferEntry<K, V>> prunedEntries) {
 			for (SharedBufferEntry<K, V> entry : entries.values()) {
 				entry.removeEdges(prunedEntries);
 			}
@@ -542,15 +542,12 @@ public void addEdge(SharedBufferEdge<K, V> edge) {
 		/**
 		 * Remove edges with the specified targets.
 		 */
-		private void removeEdges(final List<SharedBufferEntry<K, V>> prunedEntries) {
-			Iterator<SharedBufferEdge<K, V>> itor = edges.iterator();
-			while (itor.hasNext()) {
-				SharedBufferEdge<K, V> edge = itor.next();
-				for (SharedBufferEntry<K, V> prunedEntry : prunedEntries) {
-					if (prunedEntry == edge.getTarget()) {
-						itor.remove();
-						break;
-					}
+		private void removeEdges(final Set<SharedBufferEntry<K, V>> prunedEntries) {
+			Iterator<SharedBufferEdge<K, V>> it = edges.iterator();
+			while (it.hasNext()) {
+				SharedBufferEdge<K, V> edge = it.next();
+				if (prunedEntries.contains(edge.getTarget())) {
+					it.remove();
 				}
 			}
 		}