You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dl...@apache.org on 2016/06/22 02:29:32 UTC
[2/2] incubator-quarks git commit: Added keyedTimedBatchTest. Fixed
sync issue
Added keyedTimedBatchTest. Fixed sync issue
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/5e1ed08c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/5e1ed08c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/5e1ed08c
Branch: refs/heads/master
Commit: 5e1ed08ce95c2361de1c69e0f94f1457c12a1c95
Parents: e076676
Author: wcmarsha <wc...@us.ibm.com>
Authored: Tue Jun 21 17:25:39 2016 -0700
Committer: wcmarsha <wc...@us.ibm.com>
Committed: Tue Jun 21 17:25:39 2016 -0700
----------------------------------------------------------------------
.../src/main/java/quarks/window/Policies.java | 3 +-
.../java/quarks/test/window/WindowTest.java | 44 ++++++++++++++++++++
2 files changed, 46 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/5e1ed08c/api/window/src/main/java/quarks/window/Policies.java
----------------------------------------------------------------------
diff --git a/api/window/src/main/java/quarks/window/Policies.java b/api/window/src/main/java/quarks/window/Policies.java
index eebae55..6fe1876 100644
--- a/api/window/src/main/java/quarks/window/Policies.java
+++ b/api/window/src/main/java/quarks/window/Policies.java
@@ -18,6 +18,7 @@ under the License.
*/
package quarks.window;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -71,7 +72,7 @@ public class Policies {
// Can't use lambda since state is required
return new BiConsumer<Partition<T,K,L>, T>() {
- private Set<Partition<T,K,L>> initialized_partitions = new HashSet<>();
+ private Set<Partition<T,K,L>> initialized_partitions = Collections.synchronizedSet(new HashSet<>());
@Override
public void accept(Partition<T, K, L> partition, T tuple) {
if(!initialized_partitions.contains(partition)){
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/5e1ed08c/api/window/src/test/java/quarks/test/window/WindowTest.java
----------------------------------------------------------------------
diff --git a/api/window/src/test/java/quarks/test/window/WindowTest.java b/api/window/src/test/java/quarks/test/window/WindowTest.java
index f93889e..bd10dd7 100644
--- a/api/window/src/test/java/quarks/test/window/WindowTest.java
+++ b/api/window/src/test/java/quarks/test/window/WindowTest.java
@@ -383,6 +383,50 @@ public class WindowTest {
@SuppressWarnings("serial")
@Test
+ public void keyedTimeBatchWindowTest() throws InterruptedException{
+ Map<Integer, List<Long> > numBatches = Collections.synchronizedMap(new HashMap<>());
+ for(int i = 0; i < 5; i++)
+ numBatches.put(i, new LinkedList<Long>());
+
+ ScheduledExecutorService ses = new ScheduledThreadPoolExecutor(5);
+ Window<Integer, Integer, List<Integer>> window =
+ Windows.window(
+ alwaysInsert(),
+ Policies.scheduleEvictOnFirstInsert(1, TimeUnit.SECONDS),
+ Policies.evictAllAndScheduleEvictWithProcess(1, TimeUnit.SECONDS),
+ (partiton, tuple) -> {},
+ tuple -> tuple,
+ () -> new ArrayList<Integer>());
+
+ window.registerPartitionProcessor(new BiConsumer<List<Integer>, Integer>() {
+ @Override
+ public void accept(List<Integer> t, Integer u) {
+ List<Long> l = numBatches.get(u);
+ l.add((long)t.size());
+ }
+ });
+
+ window.registerScheduledExecutorService(new ScheduledThreadPoolExecutor(5));
+
+ int[] count = {1};
+ ses.scheduleAtFixedRate(() -> {
+ window.insert(count[0]++ % 5);
+ }, 0, 1, TimeUnit.MILLISECONDS);
+
+ Thread.sleep(11000);
+ double tolerance = .08;
+
+ for(Integer key : numBatches.keySet()){
+ List<Long> batch = numBatches.get(key);
+ for(Long l : batch){
+ assertTrue("Values:" + batch.toString(), withinTolerance(200.0, l.doubleValue(), tolerance));
+ }
+ }
+ }
+
+
+ @SuppressWarnings("serial")
+ @Test
public void timeBatchEnsureUnique() throws InterruptedException{
List<List<Integer>> batches = Collections.synchronizedList(new LinkedList<>());