use java.util.Map to implement Multiset

flink-core/pom.xml

@@ -80,13 +80,6 @@ under the License.
 			<!-- managed version -->
 		</dependency>
 
-		<!-- For multiset -->
-		<dependency>
-			<groupId>org.apache.commons</groupId>
-			<artifactId>commons-collections4</artifactId>
-			<version>4.1</version>
-		</dependency>
-
 		<!-- Avro is needed for the interoperability with Avro types for serialization -->
 		<dependency>
 			<groupId>org.apache.avro</groupId>

flink-core/src/main/java/org/apache/flink/api/java/typeutils/MultisetTypeInfo.java

@@ -18,12 +18,11 @@
 
 package org.apache.flink.api.java.typeutils;
 
-import org.apache.commons.collections4.multiset.AbstractMultiSet;
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.common.typeutils.base.MultisetSerializer;
+
+import java.util.Map;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -33,19 +32,17 @@
  * @param <T> The type of the elements in the Multiset.
  */
 @PublicEvolving
-public final class MultisetTypeInfo<T> extends TypeInformation<AbstractMultiSet<T>> {
+public final class MultisetTypeInfo<T> extends MapTypeInfo<T, Integer> {
 
 	private static final long serialVersionUID = 1L;
 
-	private final TypeInformation<T> elementTypeInfo;
-
 
 	public MultisetTypeInfo(Class<T> elementTypeClass) {
-		this.elementTypeInfo = of(checkNotNull(elementTypeClass, "elementTypeClass"));
+		super(elementTypeClass, Integer.class);
 	}
 
 	public MultisetTypeInfo(TypeInformation<T> elementTypeInfo) {
-		this.elementTypeInfo = checkNotNull(elementTypeInfo, "elementTypeInfo");
+		super(elementTypeInfo, BasicTypeInfo.INT_TYPE_INFO);
 	}
 
 	// ------------------------------------------------------------------------
@@ -56,7 +53,7 @@ public MultisetTypeInfo(TypeInformation<T> elementTypeInfo) {
 	 * Gets the type information for the elements contained in the Multiset
 	 */
 	public TypeInformation<T> getElementTypeInfo() {
-		return elementTypeInfo;
+		return getKeyTypeInfo();
 	}
 
 	// ------------------------------------------------------------------------
@@ -87,26 +84,20 @@ public int getTotalFields() {
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public Class<AbstractMultiSet<T>> getTypeClass() {
-		return (Class<AbstractMultiSet<T>>)(Class<?>)AbstractMultiSet.class;
+	public Class<Map<T, Integer>> getTypeClass() {
+		return (Class<Map<T, Integer>>)(Class<?>)Map.class;
 	}
 
 	@Override
 	public boolean isKeyType() {
 		return false;
 	}
 
-	@Override
-	public TypeSerializer<AbstractMultiSet<T>> createSerializer(ExecutionConfig config) {
-		TypeSerializer<T> elementTypeSerializer = elementTypeInfo.createSerializer(config);
-		return new MultisetSerializer<>(elementTypeSerializer);
-	}
-
 	// ------------------------------------------------------------------------
 
 	@Override
 	public String toString() {
-		return "Multiset<" + elementTypeInfo + '>';
+		return "Multiset<" + getKeyTypeInfo() + '>';
 	}
 
 	@Override
@@ -116,15 +107,15 @@ public boolean equals(Object obj) {
 		}
 		else if (obj instanceof MultisetTypeInfo) {
 			final MultisetTypeInfo<?> other = (MultisetTypeInfo<?>) obj;
-			return other.canEqual(this) && elementTypeInfo.equals(other.elementTypeInfo);
+			return other.canEqual(this) && getKeyTypeInfo().equals(other.getKeyTypeInfo());
 		} else {
 			return false;
 		}
 	}
 
 	@Override
 	public int hashCode() {
-		return 31 * elementTypeInfo.hashCode() + 1;
+		return 31 * getKeyTypeInfo().hashCode() + 1;
 	}
 
 	@Override