[flink] 05/10: [hotfix][runtime] Preserve singleton property of UNKNOWN ResourceSpec and ResourceProfile

[sewen]

This is an automated email from the ASF dual-hosted git repository.

sewen pushed a commit to branch release-1.9
in repository https://gitbox.apache.org/repos/asf/flink.git

commit a81344c2a15d7b8191bfd3a71a518d3cb9b7460e
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Thu Jul 11 17:11:58 2019 +0200

    [hotfix][runtime] Preserve singleton property of UNKNOWN ResourceSpec and 
ResourceProfile
---
 .../org/apache/flink/api/common/operators/ResourceSpec.java | 13 +++++++++++++
 .../apache/flink/api/common/operators/ResourceSpecTest.java |  8 ++++++++
 .../runtime/clusterframework/types/ResourceProfile.java     | 13 +++++++++++++
 .../runtime/clusterframework/types/ResourceProfileTest.java |  8 ++++++++
 4 files changed, 42 insertions(+)

diff --git 
a/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
 
b/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
index 7ddd2b3..dc6a611 100755
--- 
a/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
+++ 
b/flink-core/src/main/java/org/apache/flink/api/common/operators/ResourceSpec.java
@@ -287,6 +287,19 @@ public class ResourceSpec implements Serializable {
                                '}';
        }
 
+       // 
------------------------------------------------------------------------
+       //  serialization
+       // 
------------------------------------------------------------------------
+
+       private Object readResolve() {
+               // try to preserve the singleton property for UNKNOWN
+               return this.equals(UNKNOWN) ? UNKNOWN : this;
+       }
+
+       // 
------------------------------------------------------------------------
+       //  builder
+       // 
------------------------------------------------------------------------
+
        public static Builder newBuilder() {
                return new Builder();
        }
diff --git 
a/flink-core/src/test/java/org/apache/flink/api/common/operators/ResourceSpecTest.java
 
b/flink-core/src/test/java/org/apache/flink/api/common/operators/ResourceSpecTest.java
index 2227d89..b00be49 100755
--- 
a/flink-core/src/test/java/org/apache/flink/api/common/operators/ResourceSpecTest.java
+++ 
b/flink-core/src/test/java/org/apache/flink/api/common/operators/ResourceSpecTest.java
@@ -26,6 +26,7 @@ import org.junit.Test;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -205,4 +206,11 @@ public class ResourceSpecTest extends TestLogger {
 
                assertEquals(ResourceSpec.UNKNOWN, merged);
        }
+
+       @Test
+       public void testSingletonPropertyOfUnknown() throws Exception {
+               final ResourceSpec copiedSpec = 
CommonTestUtils.createCopySerializable(ResourceSpec.UNKNOWN);
+
+               assertSame(ResourceSpec.UNKNOWN, copiedSpec);
+       }
 }
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
index 3031b63..7ac1dd4 100755
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
@@ -444,6 +444,19 @@ public class ResourceProfile implements Serializable, 
Comparable<ResourceProfile
                        '}';
        }
 
+       // 
------------------------------------------------------------------------
+       //  serialization
+       // 
------------------------------------------------------------------------
+
+       private Object readResolve() {
+               // try to preserve the singleton property for UNKNOWN
+               return this.equals(UNKNOWN) ? UNKNOWN : this;
+       }
+
+       // 
------------------------------------------------------------------------
+       //  factories
+       // 
------------------------------------------------------------------------
+
        public static ResourceProfile fromResourceSpec(ResourceSpec 
resourceSpec, int networkMemory) {
                if (ResourceSpec.UNKNOWN.equals(resourceSpec)) {
                        return UNKNOWN;
diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/ResourceProfileTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/ResourceProfileTest.java
index ffbfb5e..5264ceb 100755
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/ResourceProfileTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/ResourceProfileTest.java
@@ -29,6 +29,7 @@ import java.util.Collections;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -257,4 +258,11 @@ public class ResourceProfileTest {
 
                
assertEquals(ResourceProfile.fromResourceSpec(ResourceSpec.UNKNOWN, 0), 
profile);
        }
+
+       @Test
+       public void testSingletonPropertyOfUnknown() throws Exception {
+               final ResourceProfile copiedProfile = 
CommonTestUtils.createCopySerializable(ResourceProfile.UNKNOWN);
+
+               assertSame(ResourceProfile.UNKNOWN, copiedProfile);
+       }
 }