You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@heron.apache.org by hu...@apache.org on 2018/07/06 21:42:04 UTC
[incubator-heron] branch master updated: Nwang/fix negative index
caused by hash mod (#2953)
This is an automated email from the ASF dual-hosted git repository.
huijun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-heron.git
The following commit(s) were added to refs/heads/master by this push:
new ae50ff3 Nwang/fix negative index caused by hash mod (#2953)
ae50ff3 is described below
commit ae50ff39e2164bed00182f2ff6ffc5755220b815
Author: Ning Wang <nw...@twitter.com>
AuthorDate: Fri Jul 6 14:42:00 2018 -0700
Nwang/fix negative index caused by hash mod (#2953)
* Fix out of boundary issue caused by .hashCode() % taskIds.size()
* update RemapCustomGrouping
* Refactor and add unit test
---
.../src/java/org/apache/heron/api/utils/Utils.java | 7 ++++
.../impl/groupings/JoinCustomGrouping.java | 5 +--
.../ReduceByKeyAndWindowCustomGrouping.java | 6 ++--
.../impl/groupings/RemapCustomGrouping.java | 3 +-
heron/api/tests/java/BUILD | 3 +-
.../java/org/apache/heron/api/utils/UtilsTest.java | 42 ++++++++++++++++++++++
.../heron/simulator/grouping/FieldsGrouping.java | 6 ++--
7 files changed, 62 insertions(+), 10 deletions(-)
diff --git a/heron/api/src/java/org/apache/heron/api/utils/Utils.java b/heron/api/src/java/org/apache/heron/api/utils/Utils.java
index 7a0eab2..e54dfdd 100644
--- a/heron/api/src/java/org/apache/heron/api/utils/Utils.java
+++ b/heron/api/src/java/org/apache/heron/api/utils/Utils.java
@@ -155,4 +155,11 @@ public final class Utils {
public static double zeroIfNaNOrInf(double x) {
return (Double.isNaN(x) || Double.isInfinite(x)) ? 0.0 : x;
}
+
+ public static Integer assignKeyToTask(int key, List<Integer> taskIds) {
+ int index = key % taskIds.size();
+ // Make sure index is not negative
+ index = index >= 0 ? index : index + taskIds.size();
+ return taskIds.get(index);
+ }
}
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/JoinCustomGrouping.java b/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/JoinCustomGrouping.java
index c1a969c..f720941 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/JoinCustomGrouping.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/JoinCustomGrouping.java
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.heron.api.grouping.CustomStreamGrouping;
import org.apache.heron.api.topology.TopologyContext;
+import org.apache.heron.api.utils.Utils;
import org.apache.heron.streamlet.SerializableFunction;
/**
@@ -52,8 +53,8 @@ public class JoinCustomGrouping<K, V> implements CustomStreamGrouping {
public List<Integer> chooseTasks(List<Object> values) {
List<Integer> ret = new ArrayList<>();
V obj = (V) values.get(0);
- int index = keyExtractor.apply(obj).hashCode() % taskIds.size();
- ret.add(taskIds.get(index));
+ int key = keyExtractor.apply(obj).hashCode();
+ ret.add(Utils.assignKeyToTask(key, taskIds));
return ret;
}
}
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/ReduceByKeyAndWindowCustomGrouping.java b/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/ReduceByKeyAndWindowCustomGrouping.java
index c8fcef2..7c228d3 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/ReduceByKeyAndWindowCustomGrouping.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/ReduceByKeyAndWindowCustomGrouping.java
@@ -25,8 +25,10 @@ import java.util.List;
import org.apache.heron.api.grouping.CustomStreamGrouping;
import org.apache.heron.api.topology.TopologyContext;
+import org.apache.heron.api.utils.Utils;
import org.apache.heron.streamlet.SerializableFunction;
+
/**
* ReduceByKeyAndWindowCustomGrouping is the class that routes the incoming tuples
* into the ReduceByKeyAndWindowOperator. It essentially ensures that the values being
@@ -54,8 +56,8 @@ public class ReduceByKeyAndWindowCustomGrouping<K, V> implements CustomStreamGro
public List<Integer> chooseTasks(List<Object> values) {
List<Integer> ret = new ArrayList<>();
V obj = (V) values.get(0);
- int index = keyExtractor.apply(obj).hashCode() % taskIds.size();
- ret.add(taskIds.get(index));
+ int key = keyExtractor.apply(obj).hashCode();
+ ret.add(Utils.assignKeyToTask(key, taskIds));
return ret;
}
}
diff --git a/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/RemapCustomGrouping.java b/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/RemapCustomGrouping.java
index c0e9378..aa71f3a 100644
--- a/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/RemapCustomGrouping.java
+++ b/heron/api/src/java/org/apache/heron/streamlet/impl/groupings/RemapCustomGrouping.java
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.heron.api.grouping.CustomStreamGrouping;
import org.apache.heron.api.topology.TopologyContext;
+import org.apache.heron.api.utils.Utils;
import org.apache.heron.streamlet.SerializableBiFunction;
/**
@@ -54,7 +55,7 @@ public class RemapCustomGrouping<R> implements CustomStreamGrouping {
R obj = (R) values.get(0);
List<Integer> targets = remapFn.apply(obj, ret.size());
for (Integer target : targets) {
- ret.add(taskIds.get(target % taskIds.size()));
+ ret.add(Utils.assignKeyToTask(target, taskIds));
}
return ret;
}
diff --git a/heron/api/tests/java/BUILD b/heron/api/tests/java/BUILD
index 971d17a..20eefa9 100644
--- a/heron/api/tests/java/BUILD
+++ b/heron/api/tests/java/BUILD
@@ -32,7 +32,8 @@ java_tests(
"org.apache.heron.streamlet.impl.operators.ReduceByKeyAndWindowOperatorTest",
"org.apache.heron.streamlet.impl.operators.GeneralReduceByKeyAndWindowOperatorTest",
"org.apache.heron.api.ConfigTest",
- "org.apache.heron.api.HeronSubmitterTest"
+ "org.apache.heron.api.HeronSubmitterTest",
+ "org.apache.heron.api.utils.UtilsTest"
],
runtime_deps = [ ":api-tests" ],
size = "small",
diff --git a/heron/api/tests/java/org/apache/heron/api/utils/UtilsTest.java b/heron/api/tests/java/org/apache/heron/api/utils/UtilsTest.java
new file mode 100644
index 0000000..add1fa1
--- /dev/null
+++ b/heron/api/tests/java/org/apache/heron/api/utils/UtilsTest.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.heron.api.utils;
+
+import java.util.ArrayList;
+
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+public class UtilsTest extends TestCase {
+
+ @Test
+ public void testAssignKeyToTask() {
+ ArrayList<Integer> taskIds = new ArrayList<Integer>();
+ taskIds.add(1);
+ taskIds.add(2);
+ assertEquals(taskIds.size(), 2);
+ assertEquals(Utils.assignKeyToTask(0, taskIds), Integer.valueOf(1));
+ assertEquals(Utils.assignKeyToTask(100, taskIds), Integer.valueOf(1));
+ assertEquals(Utils.assignKeyToTask(101, taskIds), Integer.valueOf(2));
+ assertEquals(Utils.assignKeyToTask(-100, taskIds), Integer.valueOf(1));
+ assertEquals(Utils.assignKeyToTask(-101, taskIds), Integer.valueOf(2));
+ }
+}
diff --git a/heron/simulator/src/java/org/apache/heron/simulator/grouping/FieldsGrouping.java b/heron/simulator/src/java/org/apache/heron/simulator/grouping/FieldsGrouping.java
index 86f6dac..340d34b 100644
--- a/heron/simulator/src/java/org/apache/heron/simulator/grouping/FieldsGrouping.java
+++ b/heron/simulator/src/java/org/apache/heron/simulator/grouping/FieldsGrouping.java
@@ -23,6 +23,7 @@ import java.util.LinkedList;
import java.util.List;
import org.apache.heron.api.generated.TopologyAPI;
+import org.apache.heron.api.utils.Utils;
import org.apache.heron.proto.system.HeronTuples;
public class FieldsGrouping extends Grouping {
@@ -55,10 +56,7 @@ public class FieldsGrouping extends Grouping {
taskIndex += getHashCode(tuple.getValues(indices)) % primeNumber;
}
- taskIndex = taskIndex % taskIds.size();
- // Make sure taskIndex is greater than 0
- taskIndex = taskIndex >= 0 ? taskIndex : taskIndex + taskIds.size();
- res.add(taskIds.get(taskIndex));
+ res.add(Utils.assignKeyToTask(taskIndex, taskIds));
return res;
}