[Streamlet] Fix streamlet config to apply cpu and ram in topology (#3180)
* [Streamlet] Fix streamlet config to apply cpu and ram in topology
* Set undefined cpu and ram to -1
* fix unit tests
diff --git a/heron/api/src/java/org/apache/heron/streamlet/Config.java b/heron/api/src/java/org/apache/heron/streamlet/Config.java
index f4b29c8..c9c5338 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/Config.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/Config.java
@@ -22,6 +22,7 @@
import java.io.Serializable;
+import org.apache.heron.common.basics.ByteAmount;
import org.apache.heron.streamlet.impl.KryoSerializer;
/**
@@ -33,7 +34,7 @@
public final class Config implements Serializable {
private static final long serialVersionUID = 6204498077403076352L;
private final double cpu;
- private final long ram;
+ private final ByteAmount ram;
private final DeliverySemantics deliverySemantics;
private final Serializer serializer;
private org.apache.heron.api.Config heronConfig;
@@ -61,9 +62,8 @@
private static class Defaults {
static final boolean USE_KRYO = true;
- static final org.apache.heron.api.Config CONFIG = new org.apache.heron.api.Config();
- static final double CPU = 1.0;
- static final long RAM = 100 * MB;
+ static final double CPU = -1.0; // -1 means undefined
+ static final ByteAmount RAM = ByteAmount.fromBytes(-1); // -1 means undefined
static final DeliverySemantics SEMANTICS = DeliverySemantics.ATMOST_ONCE;
static final Serializer SERIALIZER = Serializer.KRYO;
}
@@ -110,23 +110,7 @@
* @return the per-container RAM in bytes
*/
public long getPerContainerRam() {
- return ram;
- }
-
- /**
- * Gets the RAM used per topology container as a number of gigabytes
- * @return the per-container RAM in gigabytes
- */
- public long getPerContainerRamAsGigabytes() {
- return Math.round((double) ram / GB);
- }
-
- /**
- * Gets the RAM used per topology container as a number of megabytes
- * @return the per-container RAM in megabytes
- */
- public long getPerContainerRamAsMegabytes() {
- return Math.round((double) ram / MB);
+ return getPerContainerRamAsBytes();
}
/**
@@ -134,7 +118,23 @@
* @return the per-container RAM in bytes
*/
public long getPerContainerRamAsBytes() {
- return getPerContainerRam();
+ return ram.asBytes();
+ }
+
+ /**
+ * Gets the RAM used per topology container as a number of megabytes
+ * @return the per-container RAM in megabytes
+ */
+ public long getPerContainerRamAsMegabytes() {
+ return ram.asMegabytes();
+ }
+
+ /**
+ * Gets the RAM used per topology container as a number of gigabytes
+ * @return the per-container RAM in gigabytes
+ */
+ public long getPerContainerRamAsGigabytes() {
+ return ram.asGigabytes();
}
/**
@@ -170,12 +170,12 @@
public static final class Builder {
private org.apache.heron.api.Config config;
private double cpu;
- private long ram;
+ private ByteAmount ram;
private DeliverySemantics deliverySemantics;
private Serializer serializer;
private Builder() {
- config = Defaults.CONFIG;
+ config = new org.apache.heron.api.Config();
cpu = Defaults.CPU;
ram = Defaults.RAM;
deliverySemantics = Defaults.SEMANTICS;
@@ -188,6 +188,9 @@
*/
public Builder setPerContainerCpu(double perContainerCpu) {
this.cpu = perContainerCpu;
+ // Different packing algorithm might use different configs. Set all of them here.
+ config.setContainerCpuRequested(perContainerCpu);
+ config.setContainerMaxCpuHint(perContainerCpu);
return this;
}
@@ -196,8 +199,7 @@
* @param perContainerRam Per-container (per-instance) RAM expressed as a Long.
*/
public Builder setPerContainerRam(long perContainerRam) {
- this.ram = perContainerRam;
- return this;
+ return setPerContainerRamInBytes(perContainerRam);
}
/**
@@ -205,7 +207,10 @@
* @param perContainerRam Per-container (per-instance) RAM expressed as a Long.
*/
public Builder setPerContainerRamInBytes(long perContainerRam) {
- this.ram = perContainerRam;
+ this.ram = ByteAmount.fromBytes(perContainerRam);
+ // Different packing algorithm might use different configs. Set all of them here.
+ config.setContainerRamRequested(ram);
+ config.setContainerMaxRamHint(ram);
return this;
}
@@ -214,8 +219,7 @@
* @param perContainerRamMB Per-container (per-instance) RAM expressed as a Long.
*/
public Builder setPerContainerRamInMegabytes(long perContainerRamMB) {
- this.ram = perContainerRamMB * MB;
- return this;
+ return setPerContainerRam(perContainerRamMB * MB);
}
/**
@@ -223,8 +227,7 @@
* @param perContainerRamGB Per-container (per-instance) RAM expressed as a Long.
*/
public Builder setPerContainerRamInGigabytes(long perContainerRamGB) {
- this.ram = perContainerRamGB * GB;
- return this;
+ return setPerContainerRam(perContainerRamGB * GB);
}
/**
diff --git a/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java b/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java
index 345e417..ea40f01 100644
--- a/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java
+++ b/heron/api/tests/java/org/apache/heron/streamlet/impl/StreamletImplTest.java
@@ -563,12 +563,22 @@
}
@Test
- public void testConfigBuilder() {
+ public void testConfigBuilderDefaultConfig() {
Config defaultConfig = Config.defaultConfig();
assertEquals(defaultConfig.getSerializer(), Config.Serializer.KRYO);
- assertEquals(0, Double.compare(defaultConfig.getPerContainerCpu(), 1.0));
- assertEquals(defaultConfig.getPerContainerRam(), ByteAmount.fromMegabytes(100).asBytes());
+ assertEquals(0, Double.compare(defaultConfig.getPerContainerCpu(), -1.0));
+ assertEquals(defaultConfig.getPerContainerRam(), ByteAmount.fromBytes(-1).asBytes());
assertEquals(defaultConfig.getDeliverySemantics(), Config.DeliverySemantics.ATMOST_ONCE);
+
+ org.apache.heron.api.Config conf = defaultConfig.getHeronConfig();
+ assertFalse(conf.containsKey(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_CPU_REQUESTED));
+ assertFalse(conf.containsKey(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_MAX_CPU_HINT));
+ assertFalse(conf.containsKey(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_RAM_REQUESTED));
+ assertFalse(conf.containsKey(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_MAX_RAM_HINT));
+ }
+
+ @Test
+ public void testConfigBuilderNonDefaultConfig() {
Config nonDefaultConfig = Config.newBuilder()
.setDeliverySemantics(Config.DeliverySemantics.EFFECTIVELY_ONCE)
.setSerializer(Config.Serializer.JAVA)
@@ -581,6 +591,16 @@
assertEquals(nonDefaultConfig.getPerContainerRamAsGigabytes(), 10);
assertEquals(nonDefaultConfig.getPerContainerRamAsMegabytes(), 1024 * 10);
assertEquals(0, Double.compare(nonDefaultConfig.getPerContainerCpu(), 3.5));
+
+ org.apache.heron.api.Config conf = nonDefaultConfig.getHeronConfig();
+ assertEquals(conf.get(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_CPU_REQUESTED),
+ "3.5");
+ assertEquals(conf.get(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_MAX_CPU_HINT),
+ "3.5");
+ assertEquals(conf.get(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_RAM_REQUESTED),
+ "10737418240");
+ assertEquals(conf.get(org.apache.heron.api.Config.TOPOLOGY_CONTAINER_MAX_RAM_HINT),
+ "10737418240");
}
@Test
