[hotfix] Clean up AbstractID and make it immutable.

This removes the IOReadableWritable interface from AbstractID which is no longer used with IDs.

flink-core/src/main/java/org/apache/flink/util/AbstractID.java

@@ -18,37 +18,33 @@
 
 package org.apache.flink.util;
 
-import java.io.IOException;
 import java.util.Random;
 
 import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
 
 /**
  * A statistically unique identification number.
  */
 @PublicEvolving
-public class AbstractID implements IOReadableWritable, Comparable<AbstractID>, java.io.Serializable {
+public class AbstractID implements Comparable<AbstractID>, java.io.Serializable {
 
 	private static final long serialVersionUID = 1L;
-	
+
 	private static final Random RND = new Random();
-	
 
 	/** The size of a long in bytes */
 	private static final int SIZE_OF_LONG = 8;
 
 	/** The size of the ID in byte */
 	public static final int SIZE = 2 * SIZE_OF_LONG;
-	
+
+	// ------------------------------------------------------------------------
 
 	/** The upper part of the actual ID */
-	protected long upperPart;
+	protected final long upperPart;
 
 	/** The lower part of the actual ID */
-	protected long lowerPart;
+	protected final long lowerPart;
 
 	/** The memoized value returned by toString() */
 	private String toString;
@@ -79,10 +75,7 @@ public class AbstractID implements IOReadableWritable, Comparable<AbstractID>, j
 	}
 
 	/**
-	 * Creates a new abstract ID from the given one.
-	 * <p>
-	 * The given and the newly created abstract ID will be identical, i.e. a comparison by <code>equals</code> will
-	 * return <code>true</code> and both objects will have the same hash code.
+	 * Copy constructor: Creates a new abstract ID from the given one.
 	 *
 	 * @param id the abstract ID to copy
 	 */
@@ -101,7 +94,7 @@ public class AbstractID implements IOReadableWritable, Comparable<AbstractID>, j
 		this.lowerPart = RND.nextLong();
 		this.upperPart = RND.nextLong();
 	}
-	
+
 	// --------------------------------------------------------------------------------------------
 
 	/**
@@ -134,33 +127,17 @@ public class AbstractID implements IOReadableWritable, Comparable<AbstractID>, j
 		return bytes;
 	}
 
-	// --------------------------------------------------------------------------------------------
-	//  Serialization
-	// --------------------------------------------------------------------------------------------
-
-	@Override
-	public void read(DataInputView in) throws IOException {
-		this.lowerPart = in.readLong();
-		this.upperPart = in.readLong();
-
-		this.toString = null;
-	}
-
-	@Override
-	public void write(DataOutputView out) throws IOException {
-		out.writeLong(this.lowerPart);
-		out.writeLong(this.upperPart);
-	}
-
 	// --------------------------------------------------------------------------------------------
 	//  Standard Utilities
 	// --------------------------------------------------------------------------------------------
 	
 	@Override
 	public boolean equals(Object obj) {
-		if (obj != null && obj instanceof AbstractID) {
-			AbstractID src = (AbstractID) obj;
-			return src.lowerPart == this.lowerPart && src.upperPart == this.upperPart;
+		if (obj == this) {
+			return true;
+		} else if (obj != null && obj.getClass() == getClass()) {
+			AbstractID that = (AbstractID) obj;
+			return that.lowerPart == this.lowerPart && that.upperPart == this.upperPart;
 		} else {
 			return false;
 		}

flink-core/src/test/java/org/apache/flink/util/AbstractIDTest.java

@@ -18,113 +18,96 @@
 
 package org.apache.flink.util;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import org.apache.flink.core.testutils.CommonTestUtils;
 
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
 /**
  * This class contains tests for the {@link org.apache.flink.util.AbstractID} class.
  */
 public class AbstractIDTest extends TestLogger {
+
 	/**
 	 * Tests the serialization/deserialization of an abstract ID.
 	 */
 	@Test
-	public void testSerialization() {
+	public void testSerialization() throws Exception {
 		final AbstractID origID = new AbstractID();
-		try {
-			final AbstractID copyID = InstantiationUtil.createCopyWritable(origID);
+		final AbstractID copyID = CommonTestUtils.createCopySerializable(origID);
 
-			assertEquals(origID.hashCode(), copyID.hashCode());
-			assertEquals(origID, copyID);
-
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
+		assertEquals(origID.hashCode(), copyID.hashCode());
+		assertEquals(origID, copyID);
 	}
 
 	@Test
-	public void testConvertToBytes() {
-		try {
-			AbstractID origID = new AbstractID();
-
-			AbstractID copy1 = new AbstractID(origID);
-			AbstractID copy2 = new AbstractID(origID.getBytes());
-			AbstractID copy3 = new AbstractID(origID.getLowerPart(), origID.getUpperPart());
-
-			assertEquals(origID, copy1);
-			assertEquals(origID, copy2);
-			assertEquals(origID, copy3);
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
+	public void testConvertToBytes() throws Exception {
+		final AbstractID origID = new AbstractID();
+
+		AbstractID copy1 = new AbstractID(origID);
+		AbstractID copy2 = new AbstractID(origID.getBytes());
+		AbstractID copy3 = new AbstractID(origID.getLowerPart(), origID.getUpperPart());
+
+		assertEquals(origID, copy1);
+		assertEquals(origID, copy2);
+		assertEquals(origID, copy3);
 	}
 
 	@Test
-	public void testCompare() {
-		try {
-			AbstractID id1 = new AbstractID(0, 0);
-			AbstractID id2 = new AbstractID(1, 0);
-			AbstractID id3 = new AbstractID(0, 1);
-			AbstractID id4 = new AbstractID(-1, 0);
-			AbstractID id5 = new AbstractID(0, -1);
-			AbstractID id6 = new AbstractID(-1, -1);
-			
-			AbstractID id7 = new AbstractID(Long.MAX_VALUE, Long.MAX_VALUE);
-			AbstractID id8 = new AbstractID(Long.MIN_VALUE, Long.MIN_VALUE);
-			AbstractID id9 = new AbstractID(Long.MAX_VALUE, Long.MIN_VALUE);
-			AbstractID id10 = new AbstractID(Long.MIN_VALUE, Long.MAX_VALUE);
-			
-			// test self equality
-			assertEquals(0, id1.compareTo(InstantiationUtil.createCopyWritable(id1)));
-			assertEquals(0, id2.compareTo(InstantiationUtil.createCopyWritable(id2)));
-			assertEquals(0, id3.compareTo(InstantiationUtil.createCopyWritable(id3)));
-			assertEquals(0, id4.compareTo(InstantiationUtil.createCopyWritable(id4)));
-			assertEquals(0, id5.compareTo(InstantiationUtil.createCopyWritable(id5)));
-			assertEquals(0, id6.compareTo(InstantiationUtil.createCopyWritable(id6)));
-			assertEquals(0, id7.compareTo(InstantiationUtil.createCopyWritable(id7)));
-			assertEquals(0, id8.compareTo(InstantiationUtil.createCopyWritable(id8)));
-			assertEquals(0, id9.compareTo(InstantiationUtil.createCopyWritable(id9)));
-			assertEquals(0, id10.compareTo(InstantiationUtil.createCopyWritable(id10)));
-			
-			// test order
-			assertCompare(id1, id2, -1);
-			assertCompare(id1, id3, -1);
-			assertCompare(id1, id4, 1);
-			assertCompare(id1, id5, 1);
-			assertCompare(id1, id6, 1);
-			assertCompare(id2, id5, 1);
-			assertCompare(id3, id5, 1);
-			assertCompare(id2, id6, 1);
-			assertCompare(id3, id6, 1);
-			assertCompare(id1, id7, -1);
-			assertCompare(id1, id8, 1);
-			assertCompare(id7, id8, 1);
-			assertCompare(id9, id10, -1);
-			assertCompare(id7, id9, 1);
-			assertCompare(id7, id10, 1);
-			assertCompare(id8, id9, -1);
-			assertCompare(id8, id10, -1);
-		}
-		catch (Exception e) {
-			e.printStackTrace();
-			fail(e.getMessage());
-		}
+	public void testCompare() throws Exception {
+		AbstractID id1 = new AbstractID(0, 0);
+		AbstractID id2 = new AbstractID(1, 0);
+		AbstractID id3 = new AbstractID(0, 1);
+		AbstractID id4 = new AbstractID(-1, 0);
+		AbstractID id5 = new AbstractID(0, -1);
+		AbstractID id6 = new AbstractID(-1, -1);
+
+		AbstractID id7 = new AbstractID(Long.MAX_VALUE, Long.MAX_VALUE);
+		AbstractID id8 = new AbstractID(Long.MIN_VALUE, Long.MIN_VALUE);
+		AbstractID id9 = new AbstractID(Long.MAX_VALUE, Long.MIN_VALUE);
+		AbstractID id10 = new AbstractID(Long.MIN_VALUE, Long.MAX_VALUE);
+
+		// test self equality
+		assertEquals(0, id1.compareTo(CommonTestUtils.createCopySerializable(id1)));
+		assertEquals(0, id2.compareTo(CommonTestUtils.createCopySerializable(id2)));
+		assertEquals(0, id3.compareTo(CommonTestUtils.createCopySerializable(id3)));
+		assertEquals(0, id4.compareTo(CommonTestUtils.createCopySerializable(id4)));
+		assertEquals(0, id5.compareTo(CommonTestUtils.createCopySerializable(id5)));
+		assertEquals(0, id6.compareTo(CommonTestUtils.createCopySerializable(id6)));
+		assertEquals(0, id7.compareTo(CommonTestUtils.createCopySerializable(id7)));
+		assertEquals(0, id8.compareTo(CommonTestUtils.createCopySerializable(id8)));
+		assertEquals(0, id9.compareTo(CommonTestUtils.createCopySerializable(id9)));
+		assertEquals(0, id10.compareTo(CommonTestUtils.createCopySerializable(id10)));
+
+		// test order
+		assertCompare(id1, id2, -1);
+		assertCompare(id1, id3, -1);
+		assertCompare(id1, id4, 1);
+		assertCompare(id1, id5, 1);
+		assertCompare(id1, id6, 1);
+		assertCompare(id2, id5, 1);
+		assertCompare(id3, id5, 1);
+		assertCompare(id2, id6, 1);
+		assertCompare(id3, id6, 1);
+		assertCompare(id1, id7, -1);
+		assertCompare(id1, id8, 1);
+		assertCompare(id7, id8, 1);
+		assertCompare(id9, id10, -1);
+		assertCompare(id7, id9, 1);
+		assertCompare(id7, id10, 1);
+		assertCompare(id8, id9, -1);
+		assertCompare(id8, id10, -1);
 	}
-	
+
 	private static void assertCompare(AbstractID a, AbstractID b, int signum) {
 		int cmpAB = a.compareTo(b);
 		int cmpBA = b.compareTo(a);
-		
+
 		int sgnAB = cmpAB > 0 ? 1 : (cmpAB < 0 ? -1 : 0);
 		int sgnBA = cmpBA > 0 ? 1 : (cmpBA < 0 ? -1 : 0);
-		
+
 		assertEquals(signum, sgnAB);
 		assertTrue(sgnAB == -sgnBA);
 	}