You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluo.apache.org by kt...@apache.org on 2016/10/17 14:56:27 UTC
[1/2] incubator-fluo-recipes git commit: Misc release prep cleanup
Repository: incubator-fluo-recipes
Updated Branches:
refs/heads/master 573aeb788 -> 10ccb16ae
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/BigUpdateIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/BigUpdateIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/BigUpdateIT.java
deleted file mode 100644
index 3f3a84c..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/BigUpdateIT.java
+++ /dev/null
@@ -1,208 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import java.io.File;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.MapDifference;
-import com.google.common.collect.Maps;
-import org.apache.commons.io.FileUtils;
-import org.apache.fluo.api.client.FluoClient;
-import org.apache.fluo.api.client.FluoFactory;
-import org.apache.fluo.api.client.Transaction;
-import org.apache.fluo.api.client.TransactionBase;
-import org.apache.fluo.api.client.scanner.ColumnScanner;
-import org.apache.fluo.api.client.scanner.RowScanner;
-import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.ColumnValue;
-import org.apache.fluo.api.data.Span;
-import org.apache.fluo.api.mini.MiniFluo;
-import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
-import org.apache.fluo.recipes.core.types.StringEncoder;
-import org.apache.fluo.recipes.core.types.TypeLayer;
-import org.apache.fluo.recipes.core.types.TypedSnapshot;
-import org.apache.fluo.recipes.core.types.TypedTransactionBase;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-/**
- * This test configures a small buffer size and verifies that multiple passes are made to process
- * updates.
- */
-public class BigUpdateIT {
- private static final TypeLayer tl = new TypeLayer(new StringEncoder());
-
- private MiniFluo miniFluo;
-
- private CollisionFreeMap<String, Long> wcMap;
-
- static final String MAP_ID = "bu";
-
- public static class LongCombiner implements Combiner<String, Long> {
-
- @Override
- public Optional<Long> combine(String key, Iterator<Long> updates) {
- long[] count = new long[] {0};
- updates.forEachRemaining(l -> count[0] += l);
- return Optional.of(count[0]);
- }
- }
-
- static final Column DSCOL = new Column("debug", "sum");
-
- private static AtomicInteger globalUpdates = new AtomicInteger(0);
-
- public static class MyObserver extends UpdateObserver<String, Long> {
-
- @Override
- public void updatingValues(TransactionBase tx, Iterator<Update<String, Long>> updates) {
- TypedTransactionBase ttx = tl.wrap(tx);
-
- Map<String, Long> expectedOld = new HashMap<>();
-
-
- while (updates.hasNext()) {
- Update<String, Long> update = updates.next();
-
- if (update.getOldValue().isPresent()) {
- expectedOld.put("side:" + update.getKey(), update.getOldValue().get());
- }
-
- ttx.mutate().row("side:" + update.getKey()).col(DSCOL).set(update.getNewValue().get());
- }
-
- // get last values set to verify same as passed in old value
- Map<String, Long> actualOld =
- Maps.transformValues(
- ttx.get().rowsString(expectedOld.keySet()).columns(ImmutableSet.of(DSCOL))
- .toStringMap(), m -> m.get(DSCOL).toLong());
-
- MapDifference<String, Long> diff = Maps.difference(expectedOld, actualOld);
-
- Assert.assertTrue(diff.toString(), diff.areEqual());
-
- globalUpdates.incrementAndGet();
- }
- }
-
- @Before
- public void setUpFluo() throws Exception {
- FileUtils.deleteQuietly(new File("target/mini"));
-
- FluoConfiguration props = new FluoConfiguration();
- props.setApplicationName("eqt");
- props.setWorkerThreads(20);
- props.setMiniDataDir("target/mini");
-
- SimpleSerializer.setSerializer(props, TestSerializer.class);
-
- CollisionFreeMap.configure(props, new CollisionFreeMap.Options(MAP_ID, LongCombiner.class,
- MyObserver.class, String.class, Long.class, 2).setBufferSize(1 << 10));
-
- miniFluo = FluoFactory.newMiniFluo(props);
-
- wcMap = CollisionFreeMap.getInstance(MAP_ID, props.getAppConfiguration());
-
- globalUpdates.set(0);
- }
-
- @After
- public void tearDownFluo() throws Exception {
- if (miniFluo != null) {
- miniFluo.close();
- }
- }
-
- @Test
- public void testBigUpdates() {
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- updateMany(fc);
-
- miniFluo.waitForObservers();
-
- int numUpdates = 0;
-
- try (TypedSnapshot snap = tl.wrap(fc.newSnapshot())) {
- checkUpdates(snap, 1, 1000);
- numUpdates = globalUpdates.get();
- // there are two buckets, expect update processing at least twice per bucket
- Assert.assertTrue(numUpdates >= 4);
- }
-
- updateMany(fc);
- updateMany(fc);
-
- miniFluo.waitForObservers();
-
- try (TypedSnapshot snap = tl.wrap(fc.newSnapshot())) {
- checkUpdates(snap, 3, 1000);
- numUpdates = globalUpdates.get() - numUpdates;
- Assert.assertTrue(numUpdates >= 4);
- }
-
- for (int i = 0; i < 10; i++) {
- updateMany(fc);
- }
-
- miniFluo.waitForObservers();
-
- try (TypedSnapshot snap = tl.wrap(fc.newSnapshot())) {
- checkUpdates(snap, 13, 1000);
- numUpdates = globalUpdates.get() - numUpdates;
- Assert.assertTrue(numUpdates >= 4);
- }
- }
- }
-
- private void checkUpdates(TypedSnapshot snap, long expectedVal, long expectedRows) {
- RowScanner rows = snap.scanner().over(Span.prefix("side:")).byRow().build();
-
- int row = 0;
-
- for (ColumnScanner columns : rows) {
- Assert.assertEquals(String.format("side:%06d", row++), columns.getsRow());
-
- for (ColumnValue columnValue : columns) {
- Assert.assertEquals(new Column("debug", "sum"), columnValue.getColumn());
- Assert
- .assertEquals("row : " + columns.getsRow(), "" + expectedVal, columnValue.getsValue());
- }
- }
-
- Assert.assertEquals(expectedRows, row);
- }
-
- private void updateMany(FluoClient fc) {
- try (Transaction tx = fc.newTransaction()) {
- Map<String, Long> updates = new HashMap<>();
- for (int i = 0; i < 1000; i++) {
- updates.put(String.format("%06d", i), 1L);
- }
-
- wcMap.update(tx, updates);
- tx.commit();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/CollisionFreeMapIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/CollisionFreeMapIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/CollisionFreeMapIT.java
deleted file mode 100644
index f5c832a..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/CollisionFreeMapIT.java
+++ /dev/null
@@ -1,352 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import java.io.File;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Random;
-
-import com.google.common.collect.ImmutableMap;
-import org.apache.commons.io.FileUtils;
-import org.apache.fluo.api.client.FluoClient;
-import org.apache.fluo.api.client.FluoFactory;
-import org.apache.fluo.api.client.LoaderExecutor;
-import org.apache.fluo.api.client.Snapshot;
-import org.apache.fluo.api.client.Transaction;
-import org.apache.fluo.api.client.scanner.CellScanner;
-import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.ObserverSpecification;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
-import org.apache.fluo.api.data.Span;
-import org.apache.fluo.api.mini.MiniFluo;
-import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-public class CollisionFreeMapIT {
-
- private MiniFluo miniFluo;
-
- private CollisionFreeMap<String, Long> wcMap;
-
- static final String MAP_ID = "wcm";
-
- @Before
- public void setUpFluo() throws Exception {
- FileUtils.deleteQuietly(new File("target/mini"));
-
- FluoConfiguration props = new FluoConfiguration();
- props.setApplicationName("eqt");
- props.setWorkerThreads(20);
- props.setMiniDataDir("target/mini");
-
- props.addObserver(new ObserverSpecification(DocumentObserver.class.getName()));
-
- SimpleSerializer.setSerializer(props, TestSerializer.class);
-
- CollisionFreeMap.configure(props, new CollisionFreeMap.Options(MAP_ID, WordCountCombiner.class,
- WordCountObserver.class, String.class, Long.class, 17));
-
- miniFluo = FluoFactory.newMiniFluo(props);
-
- wcMap = CollisionFreeMap.getInstance(MAP_ID, props.getAppConfiguration());
- }
-
- @After
- public void tearDownFluo() throws Exception {
- if (miniFluo != null) {
- miniFluo.close();
- }
- }
-
- private Map<String, Long> getComputedWordCounts(FluoClient fc) {
- Map<String, Long> counts = new HashMap<>();
-
- try (Snapshot snap = fc.newSnapshot()) {
-
- CellScanner scanner = snap.scanner().over(Span.prefix("iwc:")).build();
-
- for (RowColumnValue rcv : scanner) {
- String[] tokens = rcv.getsRow().split(":");
- String word = tokens[2];
- Long count = Long.valueOf(tokens[1]);
-
- Assert.assertFalse("Word seen twice in index " + word, counts.containsKey(word));
-
- counts.put(word, count);
- }
- }
-
- return counts;
- }
-
- private Map<String, Long> computeWordCounts(FluoClient fc) {
- Map<String, Long> counts = new HashMap<>();
-
- try (Snapshot snap = fc.newSnapshot()) {
-
-
- CellScanner scanner =
- snap.scanner().over(Span.prefix("d:")).fetch(new Column("content", "current")).build();
-
- for (RowColumnValue rcv : scanner) {
- String[] words = rcv.getsValue().split("\\s+");
- for (String word : words) {
- if (word.isEmpty()) {
- continue;
- }
-
- counts.merge(word, 1L, Long::sum);
- }
- }
- }
-
- return counts;
- }
-
- @Test
- public void testGet() {
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- try (Transaction tx = fc.newTransaction()) {
- wcMap.update(tx, ImmutableMap.of("cat", 2L, "dog", 5L));
- tx.commit();
- }
-
- try (Transaction tx = fc.newTransaction()) {
- wcMap.update(tx, ImmutableMap.of("cat", 1L, "dog", 1L));
- tx.commit();
- }
-
- try (Transaction tx = fc.newTransaction()) {
- wcMap.update(tx, ImmutableMap.of("cat", 1L, "dog", 1L, "fish", 2L));
- tx.commit();
- }
-
- // try reading possibly before observer combines... will either see outstanding updates or a
- // current value
- try (Snapshot snap = fc.newSnapshot()) {
- Assert.assertEquals((Long) 4L, wcMap.get(snap, "cat"));
- Assert.assertEquals((Long) 7L, wcMap.get(snap, "dog"));
- Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
- }
-
- miniFluo.waitForObservers();
-
- // in this case there should be no updates, only a current value
- try (Snapshot snap = fc.newSnapshot()) {
- Assert.assertEquals((Long) 4L, wcMap.get(snap, "cat"));
- Assert.assertEquals((Long) 7L, wcMap.get(snap, "dog"));
- Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
- }
-
- Map<String, Long> expectedCounts = new HashMap<>();
- expectedCounts.put("cat", 4L);
- expectedCounts.put("dog", 7L);
- expectedCounts.put("fish", 2L);
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
-
- try (Transaction tx = fc.newTransaction()) {
- wcMap.update(tx, ImmutableMap.of("cat", 1L, "dog", -7L));
- tx.commit();
- }
-
- // there may be outstanding update and a current value for the key in this case
- try (Snapshot snap = fc.newSnapshot()) {
- Assert.assertEquals((Long) 5L, wcMap.get(snap, "cat"));
- Assert.assertNull(wcMap.get(snap, "dog"));
- Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
- }
-
- miniFluo.waitForObservers();
-
- try (Snapshot snap = fc.newSnapshot()) {
- Assert.assertEquals((Long) 5L, wcMap.get(snap, "cat"));
- Assert.assertNull(wcMap.get(snap, "dog"));
- Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
- }
-
- expectedCounts.put("cat", 5L);
- expectedCounts.remove("dog");
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
- }
- }
-
- @Test
- public void testBasic() {
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0001", "dog cat"));
- loader.execute(new DocumentLoader("0002", "cat hamster"));
- loader.execute(new DocumentLoader("0003", "milk bread cat food"));
- loader.execute(new DocumentLoader("0004", "zoo big cat"));
- }
-
- miniFluo.waitForObservers();
-
- try (Snapshot snap = fc.newSnapshot()) {
- Assert.assertEquals((Long) 4L, wcMap.get(snap, "cat"));
- Assert.assertEquals((Long) 1L, wcMap.get(snap, "milk"));
- }
-
- Map<String, Long> expectedCounts = new HashMap<>();
- expectedCounts.put("dog", 1L);
- expectedCounts.put("cat", 4L);
- expectedCounts.put("hamster", 1L);
- expectedCounts.put("milk", 1L);
- expectedCounts.put("bread", 1L);
- expectedCounts.put("food", 1L);
- expectedCounts.put("zoo", 1L);
- expectedCounts.put("big", 1L);
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0001", "dog feline"));
- }
-
- miniFluo.waitForObservers();
-
- expectedCounts.put("cat", 3L);
- expectedCounts.put("feline", 1L);
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- // swap contents of two documents... should not change doc counts
- loader.execute(new DocumentLoader("0003", "zoo big cat"));
- loader.execute(new DocumentLoader("0004", "milk bread cat food"));
- }
-
- miniFluo.waitForObservers();
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0003", "zoo big cat"));
- loader.execute(new DocumentLoader("0004", "zoo big cat"));
- }
-
- miniFluo.waitForObservers();
-
- expectedCounts.put("zoo", 2L);
- expectedCounts.put("big", 2L);
- expectedCounts.remove("milk");
- expectedCounts.remove("bread");
- expectedCounts.remove("food");
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0002", "cat cat hamster hamster"));
- }
-
- miniFluo.waitForObservers();
-
- expectedCounts.put("cat", 4L);
- expectedCounts.put("hamster", 2L);
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0002", "dog hamster"));
- }
-
- miniFluo.waitForObservers();
-
- expectedCounts.put("cat", 2L);
- expectedCounts.put("hamster", 1L);
- expectedCounts.put("dog", 2L);
-
- Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
- }
- }
-
- private static String randDocId(Random rand) {
- return String.format("%04d", rand.nextInt(5000));
- }
-
- private static String randomDocument(Random rand) {
- StringBuilder sb = new StringBuilder();
-
- String sep = "";
- for (int i = 2; i < rand.nextInt(18); i++) {
- sb.append(sep);
- sep = " ";
- sb.append(String.format("%05d", rand.nextInt(50000)));
- }
-
- return sb.toString();
- }
-
- public void diff(Map<String, Long> m1, Map<String, Long> m2) {
- for (String word : m1.keySet()) {
- Long v1 = m1.get(word);
- Long v2 = m2.get(word);
-
- if (v2 == null || !v1.equals(v2)) {
- System.out.println(word + " " + v1 + " != " + v2);
- }
- }
-
- for (String word : m2.keySet()) {
- Long v1 = m1.get(word);
- Long v2 = m2.get(word);
-
- if (v1 == null) {
- System.out.println(word + " null != " + v2);
- }
- }
- }
-
- @Test
- public void testStress() throws Exception {
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- Random rand = new Random();
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- for (int i = 0; i < 1000; i++) {
- loader.execute(new DocumentLoader(randDocId(rand), randomDocument(rand)));
- }
- }
-
- miniFluo.waitForObservers();
- assertWordCountsEqual(fc);
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- for (int i = 0; i < 100; i++) {
- loader.execute(new DocumentLoader(randDocId(rand), randomDocument(rand)));
- }
- }
-
- miniFluo.waitForObservers();
- assertWordCountsEqual(fc);
- }
- }
-
- private void assertWordCountsEqual(FluoClient fc) {
- Map<String, Long> expected = computeWordCounts(fc);
- Map<String, Long> actual = getComputedWordCounts(fc);
- if (!expected.equals(actual)) {
- diff(expected, actual);
- Assert.fail();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentLoader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentLoader.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentLoader.java
deleted file mode 100644
index 54f5ee1..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentLoader.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import org.apache.fluo.recipes.core.types.TypedLoader;
-import org.apache.fluo.recipes.core.types.TypedTransactionBase;
-
-public class DocumentLoader extends TypedLoader {
-
- String docid;
- String doc;
-
- DocumentLoader(String docid, String doc) {
- this.docid = docid;
- this.doc = doc;
- }
-
- @Override
- public void load(TypedTransactionBase tx, Context context) throws Exception {
- tx.mutate().row("d:" + docid).fam("content").qual("new").set(doc);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentObserver.java
deleted file mode 100644
index 2c79f45..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/DocumentObserver.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.common.collect.MapDifference;
-import com.google.common.collect.Maps;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.recipes.core.types.TypedObserver;
-import org.apache.fluo.recipes.core.types.TypedTransactionBase;
-
-public class DocumentObserver extends TypedObserver {
-
- CollisionFreeMap<String, Long> wcm;
-
- @Override
- public void init(Context context) throws Exception {
- wcm = CollisionFreeMap.getInstance(CollisionFreeMapIT.MAP_ID, context.getAppConfiguration());
- }
-
- @Override
- public ObservedColumn getObservedColumn() {
- return new ObservedColumn(new Column("content", "new"), NotificationType.STRONG);
- }
-
- static Map<String, Long> getWordCounts(String doc) {
- Map<String, Long> wordCounts = new HashMap<>();
- String[] words = doc.split(" ");
- for (String word : words) {
- if (word.isEmpty()) {
- continue;
- }
- wordCounts.merge(word, 1L, Long::sum);
- }
-
- return wordCounts;
- }
-
- @Override
- public void process(TypedTransactionBase tx, Bytes row, Column col) {
- String newContent = tx.get().row(row).col(col).toString();
- String currentContent = tx.get().row(row).fam("content").qual("current").toString("");
-
- Map<String, Long> newWordCounts = getWordCounts(newContent);
- Map<String, Long> currentWordCounts = getWordCounts(currentContent);
-
- Map<String, Long> changes = calculateChanges(newWordCounts, currentWordCounts);
-
- wcm.update(tx, changes);
-
- tx.mutate().row(row).fam("content").qual("current").set(newContent);
- }
-
- private static Map<String, Long> calculateChanges(Map<String, Long> newCounts,
- Map<String, Long> currCounts) {
- Map<String, Long> changes = new HashMap<>();
-
- // guava Maps class
- MapDifference<String, Long> diffs = Maps.difference(currCounts, newCounts);
-
- // compute the diffs for words that changed
- changes.putAll(Maps.transformValues(diffs.entriesDiffering(), vDiff -> vDiff.rightValue()
- - vDiff.leftValue()));
-
- // add all new words
- changes.putAll(diffs.entriesOnlyOnRight());
-
- // subtract all words no longer present
- changes.putAll(Maps.transformValues(diffs.entriesOnlyOnLeft(), l -> l * -1));
-
- return changes;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
index 37a2443..30f93ff 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/SplitsTest.java
@@ -25,6 +25,7 @@ import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.data.Bytes;
import org.apache.fluo.recipes.core.common.TableOptimizations;
import org.apache.fluo.recipes.core.map.CollisionFreeMap.Options;
+import org.apache.fluo.recipes.core.map.it.WordCountCombiner;
import org.junit.Assert;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/TestSerializer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/TestSerializer.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/TestSerializer.java
deleted file mode 100644
index b59705a..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/TestSerializer.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import org.apache.fluo.api.config.SimpleConfiguration;
-import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
-
-public class TestSerializer implements SimpleSerializer {
-
- @Override
- public <T> byte[] serialize(T obj) {
- return obj.toString().getBytes();
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public <T> T deserialize(byte[] serObj, Class<T> clazz) {
- if (clazz.equals(Long.class)) {
- return (T) Long.valueOf(new String(serObj));
- }
-
- if (clazz.equals(String.class)) {
- return (T) new String(serObj);
- }
-
- throw new IllegalArgumentException();
- }
-
- @Override
- public void init(SimpleConfiguration appConfig) {}
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountCombiner.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountCombiner.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountCombiner.java
deleted file mode 100644
index f757c10..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountCombiner.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import java.util.Iterator;
-import java.util.Optional;
-
-public class WordCountCombiner implements Combiner<String, Long> {
- @Override
- public Optional<Long> combine(String key, Iterator<Long> updates) {
- long sum = 0;
-
- while (updates.hasNext()) {
- sum += updates.next();
- }
-
- if (sum == 0) {
- return Optional.empty();
- } else {
- return Optional.of(sum);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountObserver.java
deleted file mode 100644
index 221083c..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/WordCountObserver.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.fluo.recipes.core.map;
-
-import java.util.Iterator;
-import java.util.Optional;
-
-import org.apache.fluo.api.client.TransactionBase;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-
-public class WordCountObserver extends UpdateObserver<String, Long> {
-
- @Override
- public void updatingValues(TransactionBase tx, Iterator<Update<String, Long>> updates) {
-
- while (updates.hasNext()) {
- Update<String, Long> update = updates.next();
-
- Optional<Long> oldVal = update.getOldValue();
- Optional<Long> newVal = update.getNewValue();
-
- if (oldVal.isPresent()) {
- String oldRow = String.format("iwc:%09d:%s", oldVal.get(), update.getKey());
- tx.delete(Bytes.of(oldRow), new Column(Bytes.EMPTY, Bytes.EMPTY));
- }
-
- if (newVal.isPresent()) {
- String newRow = String.format("iwc:%09d:%s", newVal.get(), update.getKey());
- tx.set(Bytes.of(newRow), new Column(Bytes.EMPTY, Bytes.EMPTY), Bytes.EMPTY);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
new file mode 100644
index 0000000..4213cff
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/BigUpdateIT.java
@@ -0,0 +1,212 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Maps;
+import org.apache.commons.io.FileUtils;
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.client.scanner.ColumnScanner;
+import org.apache.fluo.api.client.scanner.RowScanner;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.ColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.mini.MiniFluo;
+import org.apache.fluo.recipes.core.map.CollisionFreeMap;
+import org.apache.fluo.recipes.core.map.Combiner;
+import org.apache.fluo.recipes.core.map.Update;
+import org.apache.fluo.recipes.core.map.UpdateObserver;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+import org.apache.fluo.recipes.core.types.StringEncoder;
+import org.apache.fluo.recipes.core.types.TypeLayer;
+import org.apache.fluo.recipes.core.types.TypedSnapshot;
+import org.apache.fluo.recipes.core.types.TypedTransactionBase;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * This test configures a small buffer size and verifies that multiple passes are made to process
+ * updates.
+ */
+public class BigUpdateIT {
+ private static final TypeLayer tl = new TypeLayer(new StringEncoder());
+
+ private MiniFluo miniFluo;
+
+ private CollisionFreeMap<String, Long> wcMap;
+
+ static final String MAP_ID = "bu";
+
+ public static class LongCombiner implements Combiner<String, Long> {
+
+ @Override
+ public Optional<Long> combine(String key, Iterator<Long> updates) {
+ long[] count = new long[] {0};
+ updates.forEachRemaining(l -> count[0] += l);
+ return Optional.of(count[0]);
+ }
+ }
+
+ static final Column DSCOL = new Column("debug", "sum");
+
+ private static AtomicInteger globalUpdates = new AtomicInteger(0);
+
+ public static class MyObserver extends UpdateObserver<String, Long> {
+
+ @Override
+ public void updatingValues(TransactionBase tx, Iterator<Update<String, Long>> updates) {
+ TypedTransactionBase ttx = tl.wrap(tx);
+
+ Map<String, Long> expectedOld = new HashMap<>();
+
+
+ while (updates.hasNext()) {
+ Update<String, Long> update = updates.next();
+
+ if (update.getOldValue().isPresent()) {
+ expectedOld.put("side:" + update.getKey(), update.getOldValue().get());
+ }
+
+ ttx.mutate().row("side:" + update.getKey()).col(DSCOL).set(update.getNewValue().get());
+ }
+
+ // get last values set to verify same as passed in old value
+ Map<String, Long> actualOld =
+ Maps.transformValues(
+ ttx.get().rowsString(expectedOld.keySet()).columns(ImmutableSet.of(DSCOL))
+ .toStringMap(), m -> m.get(DSCOL).toLong());
+
+ MapDifference<String, Long> diff = Maps.difference(expectedOld, actualOld);
+
+ Assert.assertTrue(diff.toString(), diff.areEqual());
+
+ globalUpdates.incrementAndGet();
+ }
+ }
+
+ @Before
+ public void setUpFluo() throws Exception {
+ FileUtils.deleteQuietly(new File("target/mini"));
+
+ FluoConfiguration props = new FluoConfiguration();
+ props.setApplicationName("eqt");
+ props.setWorkerThreads(20);
+ props.setMiniDataDir("target/mini");
+
+ SimpleSerializer.setSerializer(props, TestSerializer.class);
+
+ CollisionFreeMap.configure(props, new CollisionFreeMap.Options(MAP_ID, LongCombiner.class,
+ MyObserver.class, String.class, Long.class, 2).setBufferSize(1 << 10));
+
+ miniFluo = FluoFactory.newMiniFluo(props);
+
+ wcMap = CollisionFreeMap.getInstance(MAP_ID, props.getAppConfiguration());
+
+ globalUpdates.set(0);
+ }
+
+ @After
+ public void tearDownFluo() throws Exception {
+ if (miniFluo != null) {
+ miniFluo.close();
+ }
+ }
+
+ @Test
+ public void testBigUpdates() {
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ updateMany(fc);
+
+ miniFluo.waitForObservers();
+
+ int numUpdates = 0;
+
+ try (TypedSnapshot snap = tl.wrap(fc.newSnapshot())) {
+ checkUpdates(snap, 1, 1000);
+ numUpdates = globalUpdates.get();
+ // there are two buckets, expect update processing at least twice per bucket
+ Assert.assertTrue(numUpdates >= 4);
+ }
+
+ updateMany(fc);
+ updateMany(fc);
+
+ miniFluo.waitForObservers();
+
+ try (TypedSnapshot snap = tl.wrap(fc.newSnapshot())) {
+ checkUpdates(snap, 3, 1000);
+ numUpdates = globalUpdates.get() - numUpdates;
+ Assert.assertTrue(numUpdates >= 4);
+ }
+
+ for (int i = 0; i < 10; i++) {
+ updateMany(fc);
+ }
+
+ miniFluo.waitForObservers();
+
+ try (TypedSnapshot snap = tl.wrap(fc.newSnapshot())) {
+ checkUpdates(snap, 13, 1000);
+ numUpdates = globalUpdates.get() - numUpdates;
+ Assert.assertTrue(numUpdates >= 4);
+ }
+ }
+ }
+
+ private void checkUpdates(TypedSnapshot snap, long expectedVal, long expectedRows) {
+ RowScanner rows = snap.scanner().over(Span.prefix("side:")).byRow().build();
+
+ int row = 0;
+
+ for (ColumnScanner columns : rows) {
+ Assert.assertEquals(String.format("side:%06d", row++), columns.getsRow());
+
+ for (ColumnValue columnValue : columns) {
+ Assert.assertEquals(new Column("debug", "sum"), columnValue.getColumn());
+ Assert
+ .assertEquals("row : " + columns.getsRow(), "" + expectedVal, columnValue.getsValue());
+ }
+ }
+
+ Assert.assertEquals(expectedRows, row);
+ }
+
+ private void updateMany(FluoClient fc) {
+ try (Transaction tx = fc.newTransaction()) {
+ Map<String, Long> updates = new HashMap<>();
+ for (int i = 0; i < 1000; i++) {
+ updates.put(String.format("%06d", i), 1L);
+ }
+
+ wcMap.update(tx, updates);
+ tx.commit();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
new file mode 100644
index 0000000..e22ddea
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/CollisionFreeMapIT.java
@@ -0,0 +1,353 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.commons.io.FileUtils;
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.LoaderExecutor;
+import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.scanner.CellScanner;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ObserverSpecification;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.mini.MiniFluo;
+import org.apache.fluo.recipes.core.map.CollisionFreeMap;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CollisionFreeMapIT {
+
+ private MiniFluo miniFluo;
+
+ private CollisionFreeMap<String, Long> wcMap;
+
+ static final String MAP_ID = "wcm";
+
+ @Before
+ public void setUpFluo() throws Exception {
+ FileUtils.deleteQuietly(new File("target/mini"));
+
+ FluoConfiguration props = new FluoConfiguration();
+ props.setApplicationName("eqt");
+ props.setWorkerThreads(20);
+ props.setMiniDataDir("target/mini");
+
+ props.addObserver(new ObserverSpecification(DocumentObserver.class.getName()));
+
+ SimpleSerializer.setSerializer(props, TestSerializer.class);
+
+ CollisionFreeMap.configure(props, new CollisionFreeMap.Options(MAP_ID, WordCountCombiner.class,
+ WordCountObserver.class, String.class, Long.class, 17));
+
+ miniFluo = FluoFactory.newMiniFluo(props);
+
+ wcMap = CollisionFreeMap.getInstance(MAP_ID, props.getAppConfiguration());
+ }
+
+ @After
+ public void tearDownFluo() throws Exception {
+ if (miniFluo != null) {
+ miniFluo.close();
+ }
+ }
+
+ private Map<String, Long> getComputedWordCounts(FluoClient fc) {
+ Map<String, Long> counts = new HashMap<>();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+
+ CellScanner scanner = snap.scanner().over(Span.prefix("iwc:")).build();
+
+ for (RowColumnValue rcv : scanner) {
+ String[] tokens = rcv.getsRow().split(":");
+ String word = tokens[2];
+ Long count = Long.valueOf(tokens[1]);
+
+ Assert.assertFalse("Word seen twice in index " + word, counts.containsKey(word));
+
+ counts.put(word, count);
+ }
+ }
+
+ return counts;
+ }
+
+ private Map<String, Long> computeWordCounts(FluoClient fc) {
+ Map<String, Long> counts = new HashMap<>();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+
+
+ CellScanner scanner =
+ snap.scanner().over(Span.prefix("d:")).fetch(new Column("content", "current")).build();
+
+ for (RowColumnValue rcv : scanner) {
+ String[] words = rcv.getsValue().split("\\s+");
+ for (String word : words) {
+ if (word.isEmpty()) {
+ continue;
+ }
+
+ counts.merge(word, 1L, Long::sum);
+ }
+ }
+ }
+
+ return counts;
+ }
+
+ @Test
+ public void testGet() {
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ try (Transaction tx = fc.newTransaction()) {
+ wcMap.update(tx, ImmutableMap.of("cat", 2L, "dog", 5L));
+ tx.commit();
+ }
+
+ try (Transaction tx = fc.newTransaction()) {
+ wcMap.update(tx, ImmutableMap.of("cat", 1L, "dog", 1L));
+ tx.commit();
+ }
+
+ try (Transaction tx = fc.newTransaction()) {
+ wcMap.update(tx, ImmutableMap.of("cat", 1L, "dog", 1L, "fish", 2L));
+ tx.commit();
+ }
+
+ // try reading possibly before observer combines... will either see outstanding updates or a
+ // current value
+ try (Snapshot snap = fc.newSnapshot()) {
+ Assert.assertEquals((Long) 4L, wcMap.get(snap, "cat"));
+ Assert.assertEquals((Long) 7L, wcMap.get(snap, "dog"));
+ Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
+ }
+
+ miniFluo.waitForObservers();
+
+ // in this case there should be no updates, only a current value
+ try (Snapshot snap = fc.newSnapshot()) {
+ Assert.assertEquals((Long) 4L, wcMap.get(snap, "cat"));
+ Assert.assertEquals((Long) 7L, wcMap.get(snap, "dog"));
+ Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
+ }
+
+ Map<String, Long> expectedCounts = new HashMap<>();
+ expectedCounts.put("cat", 4L);
+ expectedCounts.put("dog", 7L);
+ expectedCounts.put("fish", 2L);
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+
+ try (Transaction tx = fc.newTransaction()) {
+ wcMap.update(tx, ImmutableMap.of("cat", 1L, "dog", -7L));
+ tx.commit();
+ }
+
+ // there may be outstanding update and a current value for the key in this case
+ try (Snapshot snap = fc.newSnapshot()) {
+ Assert.assertEquals((Long) 5L, wcMap.get(snap, "cat"));
+ Assert.assertNull(wcMap.get(snap, "dog"));
+ Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
+ }
+
+ miniFluo.waitForObservers();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+ Assert.assertEquals((Long) 5L, wcMap.get(snap, "cat"));
+ Assert.assertNull(wcMap.get(snap, "dog"));
+ Assert.assertEquals((Long) 2L, wcMap.get(snap, "fish"));
+ }
+
+ expectedCounts.put("cat", 5L);
+ expectedCounts.remove("dog");
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+ }
+ }
+
+ @Test
+ public void testBasic() {
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0001", "dog cat"));
+ loader.execute(new DocumentLoader("0002", "cat hamster"));
+ loader.execute(new DocumentLoader("0003", "milk bread cat food"));
+ loader.execute(new DocumentLoader("0004", "zoo big cat"));
+ }
+
+ miniFluo.waitForObservers();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+ Assert.assertEquals((Long) 4L, wcMap.get(snap, "cat"));
+ Assert.assertEquals((Long) 1L, wcMap.get(snap, "milk"));
+ }
+
+ Map<String, Long> expectedCounts = new HashMap<>();
+ expectedCounts.put("dog", 1L);
+ expectedCounts.put("cat", 4L);
+ expectedCounts.put("hamster", 1L);
+ expectedCounts.put("milk", 1L);
+ expectedCounts.put("bread", 1L);
+ expectedCounts.put("food", 1L);
+ expectedCounts.put("zoo", 1L);
+ expectedCounts.put("big", 1L);
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0001", "dog feline"));
+ }
+
+ miniFluo.waitForObservers();
+
+ expectedCounts.put("cat", 3L);
+ expectedCounts.put("feline", 1L);
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ // swap contents of two documents... should not change doc counts
+ loader.execute(new DocumentLoader("0003", "zoo big cat"));
+ loader.execute(new DocumentLoader("0004", "milk bread cat food"));
+ }
+
+ miniFluo.waitForObservers();
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0003", "zoo big cat"));
+ loader.execute(new DocumentLoader("0004", "zoo big cat"));
+ }
+
+ miniFluo.waitForObservers();
+
+ expectedCounts.put("zoo", 2L);
+ expectedCounts.put("big", 2L);
+ expectedCounts.remove("milk");
+ expectedCounts.remove("bread");
+ expectedCounts.remove("food");
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0002", "cat cat hamster hamster"));
+ }
+
+ miniFluo.waitForObservers();
+
+ expectedCounts.put("cat", 4L);
+ expectedCounts.put("hamster", 2L);
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0002", "dog hamster"));
+ }
+
+ miniFluo.waitForObservers();
+
+ expectedCounts.put("cat", 2L);
+ expectedCounts.put("hamster", 1L);
+ expectedCounts.put("dog", 2L);
+
+ Assert.assertEquals(expectedCounts, getComputedWordCounts(fc));
+ }
+ }
+
+ private static String randDocId(Random rand) {
+ return String.format("%04d", rand.nextInt(5000));
+ }
+
+ private static String randomDocument(Random rand) {
+ StringBuilder sb = new StringBuilder();
+
+ String sep = "";
+ for (int i = 2; i < rand.nextInt(18); i++) {
+ sb.append(sep);
+ sep = " ";
+ sb.append(String.format("%05d", rand.nextInt(50000)));
+ }
+
+ return sb.toString();
+ }
+
+ public void diff(Map<String, Long> m1, Map<String, Long> m2) {
+ for (String word : m1.keySet()) {
+ Long v1 = m1.get(word);
+ Long v2 = m2.get(word);
+
+ if (v2 == null || !v1.equals(v2)) {
+ System.out.println(word + " " + v1 + " != " + v2);
+ }
+ }
+
+ for (String word : m2.keySet()) {
+ Long v1 = m1.get(word);
+ Long v2 = m2.get(word);
+
+ if (v1 == null) {
+ System.out.println(word + " null != " + v2);
+ }
+ }
+ }
+
+ @Test
+ public void testStress() throws Exception {
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ Random rand = new Random();
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ for (int i = 0; i < 1000; i++) {
+ loader.execute(new DocumentLoader(randDocId(rand), randomDocument(rand)));
+ }
+ }
+
+ miniFluo.waitForObservers();
+ assertWordCountsEqual(fc);
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ for (int i = 0; i < 100; i++) {
+ loader.execute(new DocumentLoader(randDocId(rand), randomDocument(rand)));
+ }
+ }
+
+ miniFluo.waitForObservers();
+ assertWordCountsEqual(fc);
+ }
+ }
+
+ private void assertWordCountsEqual(FluoClient fc) {
+ Map<String, Long> expected = computeWordCounts(fc);
+ Map<String, Long> actual = getComputedWordCounts(fc);
+ if (!expected.equals(actual)) {
+ diff(expected, actual);
+ Assert.fail();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentLoader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentLoader.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentLoader.java
new file mode 100644
index 0000000..c68b3b1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentLoader.java
@@ -0,0 +1,35 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import org.apache.fluo.recipes.core.types.TypedLoader;
+import org.apache.fluo.recipes.core.types.TypedTransactionBase;
+
+public class DocumentLoader extends TypedLoader {
+
+ String docid;
+ String doc;
+
+ DocumentLoader(String docid, String doc) {
+ this.docid = docid;
+ this.doc = doc;
+ }
+
+ @Override
+ public void load(TypedTransactionBase tx, Context context) throws Exception {
+ tx.mutate().row("d:" + docid).fam("content").qual("new").set(doc);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
new file mode 100644
index 0000000..c469151
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/DocumentObserver.java
@@ -0,0 +1,90 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Maps;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.recipes.core.map.CollisionFreeMap;
+import org.apache.fluo.recipes.core.types.TypedObserver;
+import org.apache.fluo.recipes.core.types.TypedTransactionBase;
+
+public class DocumentObserver extends TypedObserver {
+
+ CollisionFreeMap<String, Long> wcm;
+
+ @Override
+ public void init(Context context) throws Exception {
+ wcm = CollisionFreeMap.getInstance(CollisionFreeMapIT.MAP_ID, context.getAppConfiguration());
+ }
+
+ @Override
+ public ObservedColumn getObservedColumn() {
+ return new ObservedColumn(new Column("content", "new"), NotificationType.STRONG);
+ }
+
+ static Map<String, Long> getWordCounts(String doc) {
+ Map<String, Long> wordCounts = new HashMap<>();
+ String[] words = doc.split(" ");
+ for (String word : words) {
+ if (word.isEmpty()) {
+ continue;
+ }
+ wordCounts.merge(word, 1L, Long::sum);
+ }
+
+ return wordCounts;
+ }
+
+ @Override
+ public void process(TypedTransactionBase tx, Bytes row, Column col) {
+ String newContent = tx.get().row(row).col(col).toString();
+ String currentContent = tx.get().row(row).fam("content").qual("current").toString("");
+
+ Map<String, Long> newWordCounts = getWordCounts(newContent);
+ Map<String, Long> currentWordCounts = getWordCounts(currentContent);
+
+ Map<String, Long> changes = calculateChanges(newWordCounts, currentWordCounts);
+
+ wcm.update(tx, changes);
+
+ tx.mutate().row(row).fam("content").qual("current").set(newContent);
+ }
+
+ private static Map<String, Long> calculateChanges(Map<String, Long> newCounts,
+ Map<String, Long> currCounts) {
+ Map<String, Long> changes = new HashMap<>();
+
+ // guava Maps class
+ MapDifference<String, Long> diffs = Maps.difference(currCounts, newCounts);
+
+ // compute the diffs for words that changed
+ changes.putAll(Maps.transformValues(diffs.entriesDiffering(), vDiff -> vDiff.rightValue()
+ - vDiff.leftValue()));
+
+ // add all new words
+ changes.putAll(diffs.entriesOnlyOnRight());
+
+ // subtract all words no longer present
+ changes.putAll(Maps.transformValues(diffs.entriesOnlyOnLeft(), l -> l * -1));
+
+ return changes;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/TestSerializer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/TestSerializer.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/TestSerializer.java
new file mode 100644
index 0000000..0bc5ff6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/TestSerializer.java
@@ -0,0 +1,45 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+
+public class TestSerializer implements SimpleSerializer {
+
+ @Override
+ public <T> byte[] serialize(T obj) {
+ return obj.toString().getBytes();
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T> T deserialize(byte[] serObj, Class<T> clazz) {
+ if (clazz.equals(Long.class)) {
+ return (T) Long.valueOf(new String(serObj));
+ }
+
+ if (clazz.equals(String.class)) {
+ return (T) new String(serObj);
+ }
+
+ throw new IllegalArgumentException();
+ }
+
+ @Override
+ public void init(SimpleConfiguration appConfig) {}
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
new file mode 100644
index 0000000..df1389f
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountCombiner.java
@@ -0,0 +1,38 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import java.util.Iterator;
+import java.util.Optional;
+
+import org.apache.fluo.recipes.core.map.Combiner;
+
+public class WordCountCombiner implements Combiner<String, Long> {
+ @Override
+ public Optional<Long> combine(String key, Iterator<Long> updates) {
+ long sum = 0;
+
+ while (updates.hasNext()) {
+ sum += updates.next();
+ }
+
+ if (sum == 0) {
+ return Optional.empty();
+ } else {
+ return Optional.of(sum);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
new file mode 100644
index 0000000..1230d99
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/map/it/WordCountObserver.java
@@ -0,0 +1,49 @@
+/*
+ * 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.fluo.recipes.core.map.it;
+
+import java.util.Iterator;
+import java.util.Optional;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.recipes.core.map.Update;
+import org.apache.fluo.recipes.core.map.UpdateObserver;
+
+public class WordCountObserver extends UpdateObserver<String, Long> {
+
+ @Override
+ public void updatingValues(TransactionBase tx, Iterator<Update<String, Long>> updates) {
+
+ while (updates.hasNext()) {
+ Update<String, Long> update = updates.next();
+
+ Optional<Long> oldVal = update.getOldValue();
+ Optional<Long> newVal = update.getNewValue();
+
+ if (oldVal.isPresent()) {
+ String oldRow = String.format("iwc:%09d:%s", oldVal.get(), update.getKey());
+ tx.delete(Bytes.of(oldRow), new Column(Bytes.EMPTY, Bytes.EMPTY));
+ }
+
+ if (newVal.isPresent()) {
+ String newRow = String.format("iwc:%09d:%s", newVal.get(), update.getKey());
+ tx.set(Bytes.of(newRow), new Column(Bytes.EMPTY, Bytes.EMPTY), Bytes.EMPTY);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/kryo/pom.xml
----------------------------------------------------------------------
diff --git a/modules/kryo/pom.xml b/modules/kryo/pom.xml
index fe00f96..eaed2ee 100644
--- a/modules/kryo/pom.xml
+++ b/modules/kryo/pom.xml
@@ -17,7 +17,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.fluo</groupId>
- <artifactId>fluo-recipes-parent</artifactId>
+ <artifactId>fluo-recipes</artifactId>
<version>1.0.0-incubating-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/kryo/src/main/java/org/apache/fluo/recipes/kryo/KryoSimplerSerializer.java
----------------------------------------------------------------------
diff --git a/modules/kryo/src/main/java/org/apache/fluo/recipes/kryo/KryoSimplerSerializer.java b/modules/kryo/src/main/java/org/apache/fluo/recipes/kryo/KryoSimplerSerializer.java
index aff9e7c..e8d7997 100644
--- a/modules/kryo/src/main/java/org/apache/fluo/recipes/kryo/KryoSimplerSerializer.java
+++ b/modules/kryo/src/main/java/org/apache/fluo/recipes/kryo/KryoSimplerSerializer.java
@@ -110,7 +110,7 @@ public class KryoSimplerSerializer implements SimpleSerializer, Serializable {
public KryoSimplerSerializer() {}
/**
- * Can call this method to create a serializer w/o calling {@link #init(Configuration)}
+ * Can call this method to create a serializer w/o calling {@link #init(SimpleConfiguration)}
*/
public KryoSimplerSerializer(KryoFactory factory) {
factoryType = factory.getClass().getName();
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/kryo/src/test/java/org/apache/fluo/recipes/core/serialization/KryoSimpleSerializerTest.java
----------------------------------------------------------------------
diff --git a/modules/kryo/src/test/java/org/apache/fluo/recipes/core/serialization/KryoSimpleSerializerTest.java b/modules/kryo/src/test/java/org/apache/fluo/recipes/core/serialization/KryoSimpleSerializerTest.java
deleted file mode 100644
index 95a26a9..0000000
--- a/modules/kryo/src/test/java/org/apache/fluo/recipes/core/serialization/KryoSimpleSerializerTest.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.fluo.recipes.core.serialization;
-
-import com.esotericsoftware.kryo.pool.KryoFactory;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.recipes.kryo.KryoSimplerSerializer;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class KryoSimpleSerializerTest {
-
- private static final KryoFactory KRYO_FACTORY = new KryoSimplerSerializer.DefaultFactory();
-
- public void testColumn() {
- SimpleSerializer serializer = new KryoSimplerSerializer(KRYO_FACTORY);
- Column before = new Column("a", "b");
- byte[] barray = serializer.serialize(before);
- Column after = serializer.deserialize(barray, Column.class);
- Assert.assertEquals(before, after);
- }
-
- @Test
- public void testBytes() {
- SimpleSerializer serializer = new KryoSimplerSerializer(KRYO_FACTORY);
- Bytes before = Bytes.of("test");
- byte[] barray = serializer.serialize(before);
- Bytes after = serializer.deserialize(barray, Bytes.class);
- Assert.assertEquals(before, after);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/kryo/src/test/java/org/apache/fluo/recipes/kryo/KryoSimpleSerializerTest.java
----------------------------------------------------------------------
diff --git a/modules/kryo/src/test/java/org/apache/fluo/recipes/kryo/KryoSimpleSerializerTest.java b/modules/kryo/src/test/java/org/apache/fluo/recipes/kryo/KryoSimpleSerializerTest.java
new file mode 100644
index 0000000..d2306d2
--- /dev/null
+++ b/modules/kryo/src/test/java/org/apache/fluo/recipes/kryo/KryoSimpleSerializerTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.fluo.recipes.kryo;
+
+import com.esotericsoftware.kryo.pool.KryoFactory;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+import org.apache.fluo.recipes.kryo.KryoSimplerSerializer;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class KryoSimpleSerializerTest {
+
+ private static final KryoFactory KRYO_FACTORY = new KryoSimplerSerializer.DefaultFactory();
+
+ public void testColumn() {
+ SimpleSerializer serializer = new KryoSimplerSerializer(KRYO_FACTORY);
+ Column before = new Column("a", "b");
+ byte[] barray = serializer.serialize(before);
+ Column after = serializer.deserialize(barray, Column.class);
+ Assert.assertEquals(before, after);
+ }
+
+ @Test
+ public void testBytes() {
+ SimpleSerializer serializer = new KryoSimplerSerializer(KRYO_FACTORY);
+ Bytes before = Bytes.of("test");
+ byte[] barray = serializer.serialize(before);
+ Bytes after = serializer.deserialize(barray, Bytes.class);
+ Assert.assertEquals(before, after);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/spark/pom.xml
----------------------------------------------------------------------
diff --git a/modules/spark/pom.xml b/modules/spark/pom.xml
index 1a0969e..5cb80a4 100644
--- a/modules/spark/pom.xml
+++ b/modules/spark/pom.xml
@@ -17,7 +17,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.fluo</groupId>
- <artifactId>fluo-recipes-parent</artifactId>
+ <artifactId>fluo-recipes</artifactId>
<version>1.0.0-incubating-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/spark/src/test/java/org/apache/fluo/recipes/spark/FluoSparkHelperIT.java
----------------------------------------------------------------------
diff --git a/modules/spark/src/test/java/org/apache/fluo/recipes/spark/FluoSparkHelperIT.java b/modules/spark/src/test/java/org/apache/fluo/recipes/spark/FluoSparkHelperIT.java
deleted file mode 100644
index 4ad25e2..0000000
--- a/modules/spark/src/test/java/org/apache/fluo/recipes/spark/FluoSparkHelperIT.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * 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.fluo.recipes.spark;
-
-import java.util.List;
-
-import org.apache.fluo.api.client.FluoFactory;
-import org.apache.fluo.api.data.RowColumnValue;
-import org.apache.fluo.api.mini.MiniFluo;
-import org.apache.fluo.recipes.test.AccumuloExportITBase;
-import org.apache.fluo.recipes.test.FluoITHelper;
-import org.apache.hadoop.fs.Path;
-import org.apache.spark.api.java.JavaSparkContext;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-public class FluoSparkHelperIT extends AccumuloExportITBase {
-
- static JavaSparkContext ctx;
-
- public FluoSparkHelperIT() {
- super(false);
- }
-
- @BeforeClass
- public static void setupIT() {
- ctx = FluoSparkTestUtil.newSparkContext("fluo-spark-helper");
- }
-
- @AfterClass
- public static void teardownIT() {
- ctx.stop();
- }
-
- private List<RowColumnValue> getData() {
- return FluoITHelper.parse("arow|acf|acq|aval", "brow|bcf|bcq|bval", "crow|ccf|ccq|cval");
- }
-
- @Test
- public void testAccumuloBulkImport() throws Exception {
- FluoSparkHelper fsh =
- new FluoSparkHelper(getFluoConfiguration(), ctx.hadoopConfiguration(), new Path("/tmp/"));
- List<RowColumnValue> expected = getData();
- final String accumuloTable = "table1";
- getAccumuloConnector().tableOperations().create(accumuloTable);
- fsh.bulkImportRcvToAccumulo(FluoSparkHelper.toPairRDD(ctx.parallelize(expected)),
- accumuloTable, new FluoSparkHelper.BulkImportOptions());
- Assert.assertTrue(FluoITHelper.verifyAccumuloTable(getAccumuloConnector(), accumuloTable,
- expected));
- }
-
- @Test
- public void testFluoBulkImport() throws Exception {
- FluoSparkHelper fsh =
- new FluoSparkHelper(getFluoConfiguration(), ctx.hadoopConfiguration(), new Path("/tmp/"));
- List<RowColumnValue> expected = getData();
- fsh.bulkImportRcvToFluo(FluoSparkHelper.toPairRDD(ctx.parallelize(expected)),
- new FluoSparkHelper.BulkImportOptions());
-
- try (MiniFluo miniFluo = FluoFactory.newMiniFluo(getFluoConfiguration())) {
- Assert.assertTrue(FluoITHelper.verifyFluoTable(getFluoConfiguration(), expected));
-
- List<RowColumnValue> actualRead = FluoSparkHelper.toRcvRDD(fsh.readFromFluo(ctx)).collect();
- Assert.assertTrue(FluoITHelper.verifyRowColumnValues(expected, actualRead));
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/spark/src/test/java/org/apache/fluo/recipes/spark/it/FluoSparkHelperIT.java
----------------------------------------------------------------------
diff --git a/modules/spark/src/test/java/org/apache/fluo/recipes/spark/it/FluoSparkHelperIT.java b/modules/spark/src/test/java/org/apache/fluo/recipes/spark/it/FluoSparkHelperIT.java
new file mode 100644
index 0000000..c2d512d
--- /dev/null
+++ b/modules/spark/src/test/java/org/apache/fluo/recipes/spark/it/FluoSparkHelperIT.java
@@ -0,0 +1,84 @@
+/*
+ * 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.fluo.recipes.spark.it;
+
+import java.util.List;
+
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.mini.MiniFluo;
+import org.apache.fluo.recipes.spark.FluoSparkHelper;
+import org.apache.fluo.recipes.spark.FluoSparkTestUtil;
+import org.apache.fluo.recipes.test.AccumuloExportITBase;
+import org.apache.fluo.recipes.test.FluoITHelper;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class FluoSparkHelperIT extends AccumuloExportITBase {
+
+ static JavaSparkContext ctx;
+
+ public FluoSparkHelperIT() {
+ super(false);
+ }
+
+ @BeforeClass
+ public static void setupIT() {
+ ctx = FluoSparkTestUtil.newSparkContext("fluo-spark-helper");
+ }
+
+ @AfterClass
+ public static void teardownIT() {
+ ctx.stop();
+ }
+
+ private List<RowColumnValue> getData() {
+ return FluoITHelper.parse("arow|acf|acq|aval", "brow|bcf|bcq|bval", "crow|ccf|ccq|cval");
+ }
+
+ @Test
+ public void testAccumuloBulkImport() throws Exception {
+ FluoSparkHelper fsh =
+ new FluoSparkHelper(getFluoConfiguration(), ctx.hadoopConfiguration(), new Path("/tmp/"));
+ List<RowColumnValue> expected = getData();
+ final String accumuloTable = "table1";
+ getAccumuloConnector().tableOperations().create(accumuloTable);
+ fsh.bulkImportRcvToAccumulo(FluoSparkHelper.toPairRDD(ctx.parallelize(expected)),
+ accumuloTable, new FluoSparkHelper.BulkImportOptions());
+ Assert.assertTrue(FluoITHelper.verifyAccumuloTable(getAccumuloConnector(), accumuloTable,
+ expected));
+ }
+
+ @Test
+ public void testFluoBulkImport() throws Exception {
+ FluoSparkHelper fsh =
+ new FluoSparkHelper(getFluoConfiguration(), ctx.hadoopConfiguration(), new Path("/tmp/"));
+ List<RowColumnValue> expected = getData();
+ fsh.bulkImportRcvToFluo(FluoSparkHelper.toPairRDD(ctx.parallelize(expected)),
+ new FluoSparkHelper.BulkImportOptions());
+
+ try (MiniFluo miniFluo = FluoFactory.newMiniFluo(getFluoConfiguration())) {
+ Assert.assertTrue(FluoITHelper.verifyFluoTable(getFluoConfiguration(), expected));
+
+ List<RowColumnValue> actualRead = FluoSparkHelper.toRcvRDD(fsh.readFromFluo(ctx)).collect();
+ Assert.assertTrue(FluoITHelper.verifyRowColumnValues(expected, actualRead));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/test/pom.xml
----------------------------------------------------------------------
diff --git a/modules/test/pom.xml b/modules/test/pom.xml
index 8bfb466..86f5fbd 100644
--- a/modules/test/pom.xml
+++ b/modules/test/pom.xml
@@ -17,7 +17,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.fluo</groupId>
- <artifactId>fluo-recipes-parent</artifactId>
+ <artifactId>fluo-recipes</artifactId>
<version>1.0.0-incubating-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/test/src/main/java/org/apache/fluo/recipes/test/AccumuloExportITBase.java
----------------------------------------------------------------------
diff --git a/modules/test/src/main/java/org/apache/fluo/recipes/test/AccumuloExportITBase.java b/modules/test/src/main/java/org/apache/fluo/recipes/test/AccumuloExportITBase.java
index ba7b1e6..acc13b7 100644
--- a/modules/test/src/main/java/org/apache/fluo/recipes/test/AccumuloExportITBase.java
+++ b/modules/test/src/main/java/org/apache/fluo/recipes/test/AccumuloExportITBase.java
@@ -31,7 +31,6 @@ import org.apache.fluo.api.client.FluoFactory;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.mini.MiniFluo;
import org.apache.fluo.recipes.accumulo.ops.TableOperations;
-import org.apache.fluo.recipes.core.common.TableOptimizations;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 5c926c7..5be0541 100644
--- a/pom.xml
+++ b/pom.xml
@@ -20,8 +20,7 @@
<artifactId>fluo-parent</artifactId>
<version>1-incubating</version>
</parent>
- <groupId>org.apache.fluo</groupId>
- <artifactId>fluo-recipes-parent</artifactId>
+ <artifactId>fluo-recipes</artifactId>
<version>1.0.0-incubating-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Fluo Recipes Parent</name>
@@ -51,7 +50,7 @@
<url>https://github.com/apache/incubator-fluo-recipes/issues</url>
</issueManagement>
<properties>
- <accumulo.version>1.6.5</accumulo.version>
+ <accumulo.version>1.6.6</accumulo.version>
<curator.version>2.7.1</curator.version>
<findbugs.maxRank>13</findbugs.maxRank>
<fluo.version>1.0.0-incubating</fluo.version>
@@ -225,16 +224,6 @@
<pluginManagement>
<plugins>
<plugin>
- <groupId>org.apache.rat</groupId>
- <artifactId>apache-rat-plugin</artifactId>
- <configuration>
- <excludes>
- <exclude>README.md</exclude>
- <exclude>docs/**.md</exclude>
- </excludes>
- </configuration>
- </plugin>
- <plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-failsafe-plugin</artifactId>
<configuration>
@@ -244,29 +233,6 @@
</systemPropertyVariables>
</configuration>
</plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <archive>
- <manifest>
- <addDefaultSpecificationEntries>true</addDefaultSpecificationEntries>
- <addDefaultImplementationEntries>true</addDefaultImplementationEntries>
- </manifest>
- <manifestEntries>
- <!-- sealing breaks ITs with shaded jar, which is used by this example -->
- <Sealed>false</Sealed>
- </manifestEntries>
- </archive>
- </configuration>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-javadoc-plugin</artifactId>
- <configuration>
- <excludePackageNames>*.impl.*</excludePackageNames>
- </configuration>
- </plugin>
</plugins>
</pluginManagement>
<plugins>
[2/2] incubator-fluo-recipes git commit: Misc release prep cleanup
Posted by kt...@apache.org.
Misc release prep cleanup
Project: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/commit/10ccb16a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/tree/10ccb16a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/diff/10ccb16a
Branch: refs/heads/master
Commit: 10ccb16aedd8c141f5bd21be6822481471afa7ad
Parents: 573aeb7
Author: Keith Turner <kt...@apache.org>
Authored: Fri Oct 14 17:04:59 2016 -0400
Committer: Keith Turner <kt...@apache.org>
Committed: Fri Oct 14 17:43:06 2016 -0400
----------------------------------------------------------------------
README.md | 16 +
docs/accumulo-export-queue.md | 16 +
docs/cfm.md | 16 +
docs/export-queue.md | 16 +
docs/recording-tx.md | 16 +
docs/row-hasher.md | 16 +
docs/serialization.md | 16 +
docs/table-optimization.md | 16 +
docs/testing.md | 16 +
docs/transient.md | 16 +
modules/accumulo/pom.xml | 2 +-
modules/core/pom.xml | 2 +-
.../recipes/core/common/TableOptimizations.java | 2 +-
.../fluo/recipes/core/export/ExportBucket.java | 14 +-
.../fluo/recipes/core/impl/BucketUtil.java | 31 --
.../fluo/recipes/core/map/CollisionFreeMap.java | 22 +-
.../recipes/core/export/DocumentLoader.java | 35 --
.../recipes/core/export/DocumentObserver.java | 63 ----
.../recipes/core/export/ExportBufferIT.java | 106 ------
.../fluo/recipes/core/export/ExportQueueIT.java | 114 ------
.../recipes/core/export/ExportTestBase.java | 271 --------------
.../recipes/core/export/GsonSerializer.java | 42 ---
.../fluo/recipes/core/export/OptionsTest.java | 1 +
.../fluo/recipes/core/export/RefInfo.java | 26 --
.../fluo/recipes/core/export/RefUpdates.java | 43 ---
.../recipes/core/export/it/DocumentLoader.java | 35 ++
.../core/export/it/DocumentObserver.java | 64 ++++
.../recipes/core/export/it/ExportBufferIT.java | 107 ++++++
.../recipes/core/export/it/ExportQueueIT.java | 114 ++++++
.../recipes/core/export/it/ExportTestBase.java | 274 ++++++++++++++
.../recipes/core/export/it/GsonSerializer.java | 42 +++
.../fluo/recipes/core/export/it/RefInfo.java | 26 ++
.../fluo/recipes/core/export/it/RefUpdates.java | 43 +++
.../fluo/recipes/core/map/BigUpdateIT.java | 208 -----------
.../recipes/core/map/CollisionFreeMapIT.java | 352 ------------------
.../fluo/recipes/core/map/DocumentLoader.java | 35 --
.../fluo/recipes/core/map/DocumentObserver.java | 89 -----
.../fluo/recipes/core/map/SplitsTest.java | 1 +
.../fluo/recipes/core/map/TestSerializer.java | 45 ---
.../recipes/core/map/WordCountCombiner.java | 36 --
.../recipes/core/map/WordCountObserver.java | 47 ---
.../fluo/recipes/core/map/it/BigUpdateIT.java | 212 +++++++++++
.../recipes/core/map/it/CollisionFreeMapIT.java | 353 +++++++++++++++++++
.../recipes/core/map/it/DocumentLoader.java | 35 ++
.../recipes/core/map/it/DocumentObserver.java | 90 +++++
.../recipes/core/map/it/TestSerializer.java | 45 +++
.../recipes/core/map/it/WordCountCombiner.java | 38 ++
.../recipes/core/map/it/WordCountObserver.java | 49 +++
modules/kryo/pom.xml | 2 +-
.../recipes/kryo/KryoSimplerSerializer.java | 2 +-
.../serialization/KryoSimpleSerializerTest.java | 45 ---
.../recipes/kryo/KryoSimpleSerializerTest.java | 46 +++
modules/spark/pom.xml | 2 +-
.../fluo/recipes/spark/FluoSparkHelperIT.java | 82 -----
.../recipes/spark/it/FluoSparkHelperIT.java | 84 +++++
modules/test/pom.xml | 2 +-
.../fluo/recipes/test/AccumuloExportITBase.java | 1 -
pom.xml | 38 +-
58 files changed, 1856 insertions(+), 1722 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/README.md
----------------------------------------------------------------------
diff --git a/README.md b/README.md
index b592204..e5d3a5a 100644
--- a/README.md
+++ b/README.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Apache Fluo Recipes
[![Build Status][ti]][tl] [![Apache License][li]][ll]
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/accumulo-export-queue.md
----------------------------------------------------------------------
diff --git a/docs/accumulo-export-queue.md b/docs/accumulo-export-queue.md
index b880d58..8b4bf39 100644
--- a/docs/accumulo-export-queue.md
+++ b/docs/accumulo-export-queue.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Accumulo Export Queue Specialization
## Background
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/cfm.md
----------------------------------------------------------------------
diff --git a/docs/cfm.md b/docs/cfm.md
index 7b3a2ac..85b05b0 100644
--- a/docs/cfm.md
+++ b/docs/cfm.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Collision Free Map Recipe
## Background
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/export-queue.md
----------------------------------------------------------------------
diff --git a/docs/export-queue.md b/docs/export-queue.md
index c366e9d..7ad931f 100644
--- a/docs/export-queue.md
+++ b/docs/export-queue.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Export Queue Recipe
## Background
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/recording-tx.md
----------------------------------------------------------------------
diff --git a/docs/recording-tx.md b/docs/recording-tx.md
index 0dca693..7a6fb8e 100644
--- a/docs/recording-tx.md
+++ b/docs/recording-tx.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# RecordingTransaction recipe
A `RecordingTransaction` is an implementation of `Transaction` that logs all transaction operations
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/row-hasher.md
----------------------------------------------------------------------
diff --git a/docs/row-hasher.md b/docs/row-hasher.md
index 1f5158d..d6d603b 100644
--- a/docs/row-hasher.md
+++ b/docs/row-hasher.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Row hash prefix recipe
## Background
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/serialization.md
----------------------------------------------------------------------
diff --git a/docs/serialization.md b/docs/serialization.md
index 9b664a0..25b9d83 100644
--- a/docs/serialization.md
+++ b/docs/serialization.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Serializing Data
Various Fluo Recipes deal with POJOs and need to serialize them. The
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/table-optimization.md
----------------------------------------------------------------------
diff --git a/docs/table-optimization.md b/docs/table-optimization.md
index 7235981..f170b77 100644
--- a/docs/table-optimization.md
+++ b/docs/table-optimization.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Fluo Table Optimization
## Background
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/testing.md
----------------------------------------------------------------------
diff --git a/docs/testing.md b/docs/testing.md
index 520fd1d..0908857 100644
--- a/docs/testing.md
+++ b/docs/testing.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Testing
Fluo includes MiniFluo which makes it possible to write an integeration test that
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/docs/transient.md
----------------------------------------------------------------------
diff --git a/docs/transient.md b/docs/transient.md
index d0ac845..a73a197 100644
--- a/docs/transient.md
+++ b/docs/transient.md
@@ -1,3 +1,19 @@
+<!--
+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.
+-->
# Transient data
## Background
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/accumulo/pom.xml
----------------------------------------------------------------------
diff --git a/modules/accumulo/pom.xml b/modules/accumulo/pom.xml
index 945b9fc..751fbee 100644
--- a/modules/accumulo/pom.xml
+++ b/modules/accumulo/pom.xml
@@ -17,7 +17,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.fluo</groupId>
- <artifactId>fluo-recipes-parent</artifactId>
+ <artifactId>fluo-recipes</artifactId>
<version>1.0.0-incubating-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/pom.xml
----------------------------------------------------------------------
diff --git a/modules/core/pom.xml b/modules/core/pom.xml
index 4d6893d..720d5f0 100644
--- a/modules/core/pom.xml
+++ b/modules/core/pom.xml
@@ -17,7 +17,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.fluo</groupId>
- <artifactId>fluo-recipes-parent</artifactId>
+ <artifactId>fluo-recipes</artifactId>
<version>1.0.0-incubating-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/main/java/org/apache/fluo/recipes/core/common/TableOptimizations.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/common/TableOptimizations.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/common/TableOptimizations.java
index 3f0e5f2..f2833c6 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/common/TableOptimizations.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/common/TableOptimizations.java
@@ -77,7 +77,7 @@ public class TableOptimizations {
* initialization. After Fluo is initialized, the optimizations can be retrieved by calling
* {@link #getConfiguredOptimizations(FluoConfiguration)}.
*
- * @param application config, likely obtained from calling
+ * @param appConfig config, likely obtained from calling
* {@link FluoConfiguration#getAppConfiguration()}
* @param key A unique identifier for the optimization
* @param clazz The optimization factory type.
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
index 6dadb8c..8100b37 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportBucket.java
@@ -18,6 +18,7 @@ package org.apache.fluo.recipes.core.export;
import java.util.Iterator;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.client.scanner.CellScanner;
import org.apache.fluo.api.data.Bytes;
@@ -25,7 +26,6 @@ import org.apache.fluo.api.data.Column;
import org.apache.fluo.api.data.RowColumn;
import org.apache.fluo.api.data.RowColumnValue;
import org.apache.fluo.api.data.Span;
-import org.apache.fluo.recipes.core.impl.BucketUtil;
import org.apache.fluo.recipes.core.types.StringEncoder;
import org.apache.fluo.recipes.core.types.TypeLayer;
import org.apache.fluo.recipes.core.types.TypedTransactionBase;
@@ -47,8 +47,18 @@ class ExportBucket {
private final String qid;
private final Bytes bucketRow;
+ static String genBucketId(int bucket, int maxBucket) {
+ Preconditions.checkArgument(bucket >= 0);
+ Preconditions.checkArgument(maxBucket > 0);
+
+ int bits = 32 - Integer.numberOfLeadingZeros(maxBucket);
+ int bucketLen = bits / 4 + (bits % 4 > 0 ? 1 : 0);
+
+ return Strings.padStart(Integer.toHexString(bucket), bucketLen, '0');
+ }
+
static Bytes generateBucketRow(String qid, int bucket, int numBuckets) {
- return Bytes.of(qid + ":" + BucketUtil.genBucketId(bucket, numBuckets));
+ return Bytes.of(qid + ":" + genBucketId(bucket, numBuckets));
}
ExportBucket(TransactionBase tx, String qid, int bucket, int numBuckets) {
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/main/java/org/apache/fluo/recipes/core/impl/BucketUtil.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/impl/BucketUtil.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/impl/BucketUtil.java
deleted file mode 100644
index be28878..0000000
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/impl/BucketUtil.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.fluo.recipes.core.impl;
-
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-
-public class BucketUtil {
- public static String genBucketId(int bucket, int maxBucket) {
- Preconditions.checkArgument(bucket >= 0);
- Preconditions.checkArgument(maxBucket > 0);
-
- int bits = 32 - Integer.numberOfLeadingZeros(maxBucket);
- int bucketLen = bits / 4 + (bits % 4 > 0 ? 1 : 0);
-
- return Strings.padStart(Integer.toHexString(bucket), bucketLen, '0');
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
index d551096..715d330 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/map/CollisionFreeMap.java
@@ -29,6 +29,7 @@ import java.util.Set;
import java.util.regex.Pattern;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterators;
import com.google.common.collect.Sets;
@@ -48,7 +49,6 @@ import org.apache.fluo.recipes.core.common.TableOptimizations;
import org.apache.fluo.recipes.core.common.TableOptimizations.TableOptimizationsFactory;
import org.apache.fluo.recipes.core.common.RowRange;
import org.apache.fluo.recipes.core.common.TransientRegistry;
-import org.apache.fluo.recipes.core.impl.BucketUtil;
import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
/**
@@ -290,7 +290,7 @@ public class CollisionFreeMap<K, V> {
byte[] k = serializer.serialize(key);
int hash = Hashing.murmur3_32().hashBytes(k).asInt();
- String bucketId = BucketUtil.genBucketId(Math.abs(hash % numBuckets), numBuckets);
+ String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
BytesBuilder rowBuilder = Bytes.builder();
@@ -352,7 +352,7 @@ public class CollisionFreeMap<K, V> {
for (Entry<K, V> entry : updates.entrySet()) {
byte[] k = serializer.serialize(entry.getKey());
int hash = Hashing.murmur3_32().hashBytes(k).asInt();
- String bucketId = BucketUtil.genBucketId(Math.abs(hash % numBuckets), numBuckets);
+ String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
// reset to the common row prefix
rowBuilder.setLength(prefixLength);
@@ -377,6 +377,16 @@ public class CollisionFreeMap<K, V> {
}
}
+ static String genBucketId(int bucket, int maxBucket) {
+ Preconditions.checkArgument(bucket >= 0);
+ Preconditions.checkArgument(maxBucket > 0);
+
+ int bits = 32 - Integer.numberOfLeadingZeros(maxBucket);
+ int bucketLen = bits / 4 + (bits % 4 > 0 ? 1 : 0);
+
+ return Strings.padStart(Integer.toHexString(bucket), bucketLen, '0');
+ }
+
public static <K2, V2> CollisionFreeMap<K2, V2> getInstance(String mapId,
SimpleConfiguration appConf) {
Options opts = new Options(mapId, appConf);
@@ -424,7 +434,7 @@ public class CollisionFreeMap<K, V> {
public RowColumnValue convert(K2 key, V2 val) {
byte[] k = serializer.serialize(key);
int hash = Hashing.murmur3_32().hashBytes(k).asInt();
- String bucketId = BucketUtil.genBucketId(Math.abs(hash % numBuckets), numBuckets);
+ String bucketId = genBucketId(Math.abs(hash % numBuckets), numBuckets);
BytesBuilder bb = Bytes.builder();
Bytes row = bb.append(mapId).append(":d:").append(bucketId).append(":").append(k).toBytes();
@@ -605,7 +615,7 @@ public class CollisionFreeMap<K, V> {
List<Bytes> dataSplits = new ArrayList<>();
for (int i = opts.getBucketsPerTablet(); i < opts.numBuckets; i += opts.getBucketsPerTablet()) {
- String bucketId = BucketUtil.genBucketId(i, opts.numBuckets);
+ String bucketId = genBucketId(i, opts.numBuckets);
rowBuilder.setLength(mapId.length());
dataSplits.add(rowBuilder.append(":d:").append(bucketId).toBytes());
}
@@ -613,7 +623,7 @@ public class CollisionFreeMap<K, V> {
List<Bytes> updateSplits = new ArrayList<>();
for (int i = opts.getBucketsPerTablet(); i < opts.numBuckets; i += opts.getBucketsPerTablet()) {
- String bucketId = BucketUtil.genBucketId(i, opts.numBuckets);
+ String bucketId = genBucketId(i, opts.numBuckets);
rowBuilder.setLength(mapId.length());
updateSplits.add(rowBuilder.append(":u:").append(bucketId).toBytes());
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentLoader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentLoader.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentLoader.java
deleted file mode 100644
index f7af791..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentLoader.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import org.apache.fluo.recipes.core.types.TypedLoader;
-import org.apache.fluo.recipes.core.types.TypedTransactionBase;
-
-public class DocumentLoader extends TypedLoader {
-
- String docid;
- String refs[];
-
- DocumentLoader(String docid, String... refs) {
- this.docid = docid;
- this.refs = refs;
- }
-
- @Override
- public void load(TypedTransactionBase tx, Context context) throws Exception {
- tx.mutate().row("d:" + docid).fam("content").qual("new").set(String.join(" ", refs));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentObserver.java
deleted file mode 100644
index c4c11d8..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/DocumentObserver.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.Set;
-
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.recipes.core.export.ExportTestBase.RefExporter;
-import org.apache.fluo.recipes.core.types.TypedObserver;
-import org.apache.fluo.recipes.core.types.TypedTransactionBase;
-
-public class DocumentObserver extends TypedObserver {
-
- ExportQueue<String, RefUpdates> refExportQueue;
-
- @Override
- public void init(Context context) throws Exception {
- refExportQueue = ExportQueue.getInstance(RefExporter.QUEUE_ID, context.getAppConfiguration());
- }
-
- @Override
- public ObservedColumn getObservedColumn() {
- return new ObservedColumn(new Column("content", "new"), NotificationType.STRONG);
- }
-
- @Override
- public void process(TypedTransactionBase tx, Bytes row, Column col) {
- String newContent = tx.get().row(row).col(col).toString();
- Set<String> newRefs = new HashSet<>(Arrays.asList(newContent.split(" ")));
- Set<String> currentRefs =
- new HashSet<>(Arrays.asList(tx.get().row(row).fam("content").qual("current").toString("")
- .split(" ")));
-
- Set<String> addedRefs = new HashSet<>(newRefs);
- addedRefs.removeAll(currentRefs);
-
- Set<String> deletedRefs = new HashSet<>(currentRefs);
- deletedRefs.removeAll(newRefs);
-
- String key = row.toString().substring(2);
- RefUpdates val = new RefUpdates(addedRefs, deletedRefs);
-
- refExportQueue.add(tx, key, val);
-
- tx.mutate().row(row).fam("content").qual("current").set(newContent);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportBufferIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportBufferIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportBufferIT.java
deleted file mode 100644
index d54982b..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportBufferIT.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-
-import org.apache.fluo.api.client.FluoClient;
-import org.apache.fluo.api.client.FluoFactory;
-import org.apache.fluo.api.client.Transaction;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class ExportBufferIT extends ExportTestBase {
-
- @Override
- protected int getNumBuckets() {
- return 2;
- }
-
- @Override
- protected Integer getBufferSize() {
- return 1024;
- }
-
- @Test
- public void testSmallExportBuffer() {
- // try setting the export buffer size small. Make sure everything is exported.
-
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- ExportQueue<String, RefUpdates> refExportQueue =
- ExportQueue.getInstance(RefExporter.QUEUE_ID, fc.getAppConfiguration());
- try (Transaction tx = fc.newTransaction()) {
- for (int i = 0; i < 1000; i++) {
- refExportQueue.add(tx, nk(i), new RefUpdates(ns(i + 10, i + 20), ns(new int[0])));
- }
-
- tx.commit();
- }
- }
-
- miniFluo.waitForObservers();
-
- Map<String, Set<String>> erefs = getExportedReferees();
- Map<String, Set<String>> expected = new HashMap<>();
-
- for (int i = 0; i < 1000; i++) {
- expected.computeIfAbsent(nk(i + 10), s -> new HashSet<>()).add(nk(i));
- expected.computeIfAbsent(nk(i + 20), s -> new HashSet<>()).add(nk(i));
- }
-
- assertEquals(expected, erefs);
- int prevNumExportCalls = getNumExportCalls();
- Assert.assertTrue(prevNumExportCalls > 10); // with small buffer there should be lots of exports
- // calls
-
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- ExportQueue<String, RefUpdates> refExportQueue =
- ExportQueue.getInstance(RefExporter.QUEUE_ID, fc.getAppConfiguration());
- try (Transaction tx = fc.newTransaction()) {
- for (int i = 0; i < 1000; i++) {
- refExportQueue.add(tx, nk(i), new RefUpdates(ns(i + 12), ns(i + 10)));
- }
-
- tx.commit();
- }
- }
-
- miniFluo.waitForObservers();
-
- erefs = getExportedReferees();
- expected = new HashMap<>();
-
- for (int i = 0; i < 1000; i++) {
- expected.computeIfAbsent(nk(i + 12), s -> new HashSet<>()).add(nk(i));
- expected.computeIfAbsent(nk(i + 20), s -> new HashSet<>()).add(nk(i));
- }
-
- assertEquals(expected, erefs);
- prevNumExportCalls = getNumExportCalls() - prevNumExportCalls;
- Assert.assertTrue(prevNumExportCalls > 10);
- }
-
- public void assertEquals(Map<String, Set<String>> expected, Map<String, Set<String>> actual) {
- if (!expected.equals(actual)) {
- System.out.println("*** diff ***");
- diff(expected, actual);
- Assert.fail();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportQueueIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportQueueIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportQueueIT.java
deleted file mode 100644
index b4e167c..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportQueueIT.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import org.apache.fluo.api.client.FluoClient;
-import org.apache.fluo.api.client.FluoFactory;
-import org.apache.fluo.api.client.LoaderExecutor;
-import org.apache.fluo.api.config.FluoConfiguration;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class ExportQueueIT extends ExportTestBase {
-
- @Test
- public void testExport() {
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0999", "0005", "0002"));
- loader.execute(new DocumentLoader("0002", "0999", "0042"));
- loader.execute(new DocumentLoader("0005", "0999", "0042"));
- loader.execute(new DocumentLoader("0042", "0999"));
- }
-
- miniFluo.waitForObservers();
-
- Assert.assertEquals(ns("0002", "0005", "0042"), getExportedReferees("0999"));
- Assert.assertEquals(ns("0999"), getExportedReferees("0002"));
- Assert.assertEquals(ns("0999"), getExportedReferees("0005"));
- Assert.assertEquals(ns("0002", "0005"), getExportedReferees("0042"));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0999", "0005", "0042"));
- }
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0999", "0005"));
- }
-
- miniFluo.waitForObservers();
-
- Assert.assertEquals(ns("0002", "0005", "0042"), getExportedReferees("0999"));
- Assert.assertEquals(ns(new String[0]), getExportedReferees("0002"));
- Assert.assertEquals(ns("0999"), getExportedReferees("0005"));
- Assert.assertEquals(ns("0002", "0005"), getExportedReferees("0042"));
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0042", "0999", "0002", "0005"));
- loader.execute(new DocumentLoader("0005", "0002"));
- }
-
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- loader.execute(new DocumentLoader("0005", "0003"));
- }
-
- miniFluo.waitForObservers();
-
- Assert.assertEquals(ns("0002", "0042"), getExportedReferees("0999"));
- Assert.assertEquals(ns("0042"), getExportedReferees("0002"));
- Assert.assertEquals(ns("0005"), getExportedReferees("0003"));
- Assert.assertEquals(ns("0999", "0042"), getExportedReferees("0005"));
- Assert.assertEquals(ns("0002"), getExportedReferees("0042"));
-
- }
- }
-
- @Test
- public void exportStressTest() {
- FluoConfiguration config = new FluoConfiguration(miniFluo.getClientConfiguration());
- config.setLoaderQueueSize(100);
- config.setLoaderThreads(20);
-
- try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
-
- loadRandom(fc, 1000, 500);
-
- miniFluo.waitForObservers();
-
- diff(getFluoReferees(fc), getExportedReferees());
-
- assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
-
- loadRandom(fc, 1000, 500);
-
- miniFluo.waitForObservers();
-
- assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
-
- loadRandom(fc, 1000, 10000);
-
- miniFluo.waitForObservers();
-
- assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
-
- loadRandom(fc, 1000, 10000);
-
- miniFluo.waitForObservers();
-
- assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportTestBase.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportTestBase.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportTestBase.java
deleted file mode 100644
index fd3ece1..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/ExportTestBase.java
+++ /dev/null
@@ -1,271 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import java.io.File;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
-
-import com.google.common.collect.Iterators;
-import org.apache.commons.io.FileUtils;
-import org.apache.fluo.api.client.FluoClient;
-import org.apache.fluo.api.client.FluoFactory;
-import org.apache.fluo.api.client.LoaderExecutor;
-import org.apache.fluo.api.client.Snapshot;
-import org.apache.fluo.api.client.scanner.CellScanner;
-import org.apache.fluo.api.client.scanner.ColumnScanner;
-import org.apache.fluo.api.client.scanner.RowScanner;
-import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.ObserverSpecification;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.ColumnValue;
-import org.apache.fluo.api.data.Span;
-import org.apache.fluo.api.mini.MiniFluo;
-import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-
-public class ExportTestBase {
-
- private static Map<String, Map<String, RefInfo>> globalExports = new HashMap<>();
- private static int exportCalls = 0;
-
- protected static Set<String> getExportedReferees(String node) {
- synchronized (globalExports) {
- Set<String> ret = new HashSet<>();
-
- Map<String, RefInfo> referees = globalExports.get(node);
-
- if (referees == null) {
- return ret;
- }
-
- referees.forEach((k, v) -> {
- if (!v.deleted)
- ret.add(k);
- });
-
- return ret;
- }
- }
-
- protected static Map<String, Set<String>> getExportedReferees() {
- synchronized (globalExports) {
-
- Map<String, Set<String>> ret = new HashMap<>();
-
- for (String k : globalExports.keySet()) {
- Set<String> referees = getExportedReferees(k);
- if (referees.size() > 0) {
- ret.put(k, referees);
- }
- }
-
- return ret;
- }
- }
-
- protected static int getNumExportCalls() {
- synchronized (globalExports) {
- return exportCalls;
- }
- }
-
- public static class RefExporter extends Exporter<String, RefUpdates> {
-
- public static final String QUEUE_ID = "req";
-
- private void updateExports(String key, long seq, String addedRef, boolean deleted) {
- Map<String, RefInfo> referees = globalExports.computeIfAbsent(addedRef, k -> new HashMap<>());
- referees.compute(key, (k, v) -> (v == null || v.seq < seq) ? new RefInfo(seq, deleted) : v);
- }
-
- @Override
- protected void processExports(Iterator<SequencedExport<String, RefUpdates>> exportIterator) {
- ArrayList<SequencedExport<String, RefUpdates>> exportList = new ArrayList<>();
- Iterators.addAll(exportList, exportIterator);
-
- synchronized (globalExports) {
- exportCalls++;
-
- for (SequencedExport<String, RefUpdates> se : exportList) {
- for (String addedRef : se.getValue().getAddedRefs()) {
- updateExports(se.getKey(), se.getSequence(), addedRef, false);
- }
-
- for (String deletedRef : se.getValue().getDeletedRefs()) {
- updateExports(se.getKey(), se.getSequence(), deletedRef, true);
- }
- }
- }
- }
- }
-
- protected MiniFluo miniFluo;
-
- protected int getNumBuckets() {
- return 13;
- }
-
- protected Integer getBufferSize() {
- return null;
- }
-
- @Before
- public void setUpFluo() throws Exception {
- FileUtils.deleteQuietly(new File("target/mini"));
-
- FluoConfiguration props = new FluoConfiguration();
- props.setApplicationName("eqt");
- props.setWorkerThreads(20);
- props.setMiniDataDir("target/mini");
-
- ObserverSpecification doc = new ObserverSpecification(DocumentObserver.class.getName());
- props.addObserver(doc);
-
- SimpleSerializer.setSerializer(props, GsonSerializer.class);
-
- ExportQueue.Options exportQueueOpts =
- new ExportQueue.Options(RefExporter.QUEUE_ID, RefExporter.class, String.class,
- RefUpdates.class, getNumBuckets());
-
- if (getBufferSize() != null) {
- exportQueueOpts.setBufferSize(getBufferSize());
- }
-
- ExportQueue.configure(props, exportQueueOpts);
-
- miniFluo = FluoFactory.newMiniFluo(props);
-
- globalExports.clear();
- exportCalls = 0;
- }
-
- @After
- public void tearDownFluo() throws Exception {
- if (miniFluo != null) {
- miniFluo.close();
- }
- }
-
- protected static Set<String> ns(String... sa) {
- return new HashSet<>(Arrays.asList(sa));
- }
-
- protected static String nk(int i) {
- return String.format("%06d", i);
- }
-
- protected static Set<String> ns(int... ia) {
- HashSet<String> ret = new HashSet<>();
- for (int i : ia) {
- ret.add(nk(i));
- }
- return ret;
- }
-
- public void assertEquals(Map<String, Set<String>> expected, Map<String, Set<String>> actual,
- FluoClient fc) {
- if (!expected.equals(actual)) {
- System.out.println("*** diff ***");
- diff(expected, actual);
- System.out.println("*** fluo dump ***");
- dump(fc);
- System.out.println("*** map dump ***");
-
- Assert.fail();
- }
- }
-
- protected void loadRandom(FluoClient fc, int num, int maxDocId) {
- try (LoaderExecutor loader = fc.newLoaderExecutor()) {
- Random rand = new Random();
-
- for (int i = 0; i < num; i++) {
- String docid = String.format("%05d", rand.nextInt(maxDocId));
- String[] refs = new String[rand.nextInt(20) + 1];
- for (int j = 0; j < refs.length; j++) {
- refs[j] = String.format("%05d", rand.nextInt(maxDocId));
- }
-
- loader.execute(new DocumentLoader(docid, refs));
- }
- }
- }
-
- protected void diff(Map<String, Set<String>> fr, Map<String, Set<String>> er) {
- HashSet<String> allKeys = new HashSet<>(fr.keySet());
- allKeys.addAll(er.keySet());
-
- for (String k : allKeys) {
- Set<String> s1 = fr.getOrDefault(k, Collections.emptySet());
- Set<String> s2 = er.getOrDefault(k, Collections.emptySet());
-
- HashSet<String> sub1 = new HashSet<>(s1);
- sub1.removeAll(s2);
-
- HashSet<String> sub2 = new HashSet<>(s2);
- sub2.removeAll(s1);
-
- if (sub1.size() > 0 || sub2.size() > 0) {
- System.out.println(k + " " + sub1 + " " + sub2);
- }
-
- }
- }
-
- protected Map<String, Set<String>> getFluoReferees(FluoClient fc) {
- Map<String, Set<String>> fluoReferees = new HashMap<>();
-
- try (Snapshot snap = fc.newSnapshot()) {
-
- Column currCol = new Column("content", "current");
- RowScanner rowScanner = snap.scanner().over(Span.prefix("d:")).fetch(currCol).byRow().build();
-
- for (ColumnScanner columnScanner : rowScanner) {
- String docid = columnScanner.getsRow().substring(2);
-
- for (ColumnValue columnValue : columnScanner) {
- String[] refs = columnValue.getsValue().split(" ");
-
- for (String ref : refs) {
- if (ref.isEmpty())
- continue;
-
- fluoReferees.computeIfAbsent(ref, k -> new HashSet<>()).add(docid);
- }
- }
- }
- }
- return fluoReferees;
- }
-
- public static void dump(FluoClient fc) {
- try (Snapshot snap = fc.newSnapshot()) {
- CellScanner scanner = snap.scanner().build();
- scanner.forEach(rcv -> System.out.println("row:[" + rcv.getRow() + "] col:["
- + rcv.getColumn() + "] val:[" + rcv.getValue() + "]"));
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/GsonSerializer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/GsonSerializer.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/GsonSerializer.java
deleted file mode 100644
index 2d45ff3..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/GsonSerializer.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import java.nio.charset.StandardCharsets;
-
-import com.google.gson.Gson;
-import org.apache.fluo.api.config.SimpleConfiguration;
-import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
-
-public class GsonSerializer implements SimpleSerializer {
-
- private Gson gson = new Gson();
-
- @Override
- public void init(SimpleConfiguration appConfig) {
-
- }
-
- @Override
- public <T> byte[] serialize(T obj) {
- return gson.toJson(obj).getBytes(StandardCharsets.UTF_8);
- }
-
- @Override
- public <T> T deserialize(byte[] serObj, Class<T> clazz) {
- return gson.fromJson(new String(serObj, StandardCharsets.UTF_8), clazz);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
index 60982bd..376223b 100644
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/OptionsTest.java
@@ -17,6 +17,7 @@ package org.apache.fluo.recipes.core.export;
import org.apache.fluo.api.config.FluoConfiguration;
import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.recipes.core.export.ExportQueue;
import org.apache.fluo.recipes.core.export.ExportQueue.Options;
import org.junit.Assert;
import org.junit.Test;
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefInfo.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefInfo.java
deleted file mode 100644
index f4d1c76..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefInfo.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-class RefInfo {
- long seq;
- boolean deleted;
-
- public RefInfo(long seq, boolean deleted) {
- this.seq = seq;
- this.deleted = deleted;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefUpdates.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefUpdates.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefUpdates.java
deleted file mode 100644
index efa94dd..0000000
--- a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/RefUpdates.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.fluo.recipes.core.export;
-
-import java.util.Set;
-
-public class RefUpdates {
- private Set<String> addedRefs;
- private Set<String> deletedRefs;
-
- public RefUpdates() {}
-
- public RefUpdates(Set<String> addedRefs, Set<String> deletedRefs) {
- this.addedRefs = addedRefs;
- this.deletedRefs = deletedRefs;
- }
-
- public Set<String> getAddedRefs() {
- return addedRefs;
- }
-
- public Set<String> getDeletedRefs() {
- return deletedRefs;
- }
-
- @Override
- public String toString() {
- return "added:" + addedRefs + " deleted:" + deletedRefs;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentLoader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentLoader.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentLoader.java
new file mode 100644
index 0000000..6b8ca36
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentLoader.java
@@ -0,0 +1,35 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+import org.apache.fluo.recipes.core.types.TypedLoader;
+import org.apache.fluo.recipes.core.types.TypedTransactionBase;
+
+public class DocumentLoader extends TypedLoader {
+
+ String docid;
+ String refs[];
+
+ DocumentLoader(String docid, String... refs) {
+ this.docid = docid;
+ this.refs = refs;
+ }
+
+ @Override
+ public void load(TypedTransactionBase tx, Context context) throws Exception {
+ tx.mutate().row("d:" + docid).fam("content").qual("new").set(String.join(" ", refs));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
new file mode 100644
index 0000000..c9765a7
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/DocumentObserver.java
@@ -0,0 +1,64 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.recipes.core.export.ExportQueue;
+import org.apache.fluo.recipes.core.export.it.ExportTestBase.RefExporter;
+import org.apache.fluo.recipes.core.types.TypedObserver;
+import org.apache.fluo.recipes.core.types.TypedTransactionBase;
+
+public class DocumentObserver extends TypedObserver {
+
+ ExportQueue<String, RefUpdates> refExportQueue;
+
+ @Override
+ public void init(Context context) throws Exception {
+ refExportQueue = ExportQueue.getInstance(RefExporter.QUEUE_ID, context.getAppConfiguration());
+ }
+
+ @Override
+ public ObservedColumn getObservedColumn() {
+ return new ObservedColumn(new Column("content", "new"), NotificationType.STRONG);
+ }
+
+ @Override
+ public void process(TypedTransactionBase tx, Bytes row, Column col) {
+ String newContent = tx.get().row(row).col(col).toString();
+ Set<String> newRefs = new HashSet<>(Arrays.asList(newContent.split(" ")));
+ Set<String> currentRefs =
+ new HashSet<>(Arrays.asList(tx.get().row(row).fam("content").qual("current").toString("")
+ .split(" ")));
+
+ Set<String> addedRefs = new HashSet<>(newRefs);
+ addedRefs.removeAll(currentRefs);
+
+ Set<String> deletedRefs = new HashSet<>(currentRefs);
+ deletedRefs.removeAll(newRefs);
+
+ String key = row.toString().substring(2);
+ RefUpdates val = new RefUpdates(addedRefs, deletedRefs);
+
+ refExportQueue.add(tx, key, val);
+
+ tx.mutate().row(row).fam("content").qual("current").set(newContent);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportBufferIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportBufferIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportBufferIT.java
new file mode 100644
index 0000000..d937ac8
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportBufferIT.java
@@ -0,0 +1,107 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.recipes.core.export.ExportQueue;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ExportBufferIT extends ExportTestBase {
+
+ @Override
+ protected int getNumBuckets() {
+ return 2;
+ }
+
+ @Override
+ protected Integer getBufferSize() {
+ return 1024;
+ }
+
+ @Test
+ public void testSmallExportBuffer() {
+ // try setting the export buffer size small. Make sure everything is exported.
+
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ ExportQueue<String, RefUpdates> refExportQueue =
+ ExportQueue.getInstance(RefExporter.QUEUE_ID, fc.getAppConfiguration());
+ try (Transaction tx = fc.newTransaction()) {
+ for (int i = 0; i < 1000; i++) {
+ refExportQueue.add(tx, nk(i), new RefUpdates(ns(i + 10, i + 20), ns(new int[0])));
+ }
+
+ tx.commit();
+ }
+ }
+
+ miniFluo.waitForObservers();
+
+ Map<String, Set<String>> erefs = getExportedReferees();
+ Map<String, Set<String>> expected = new HashMap<>();
+
+ for (int i = 0; i < 1000; i++) {
+ expected.computeIfAbsent(nk(i + 10), s -> new HashSet<>()).add(nk(i));
+ expected.computeIfAbsent(nk(i + 20), s -> new HashSet<>()).add(nk(i));
+ }
+
+ assertEquals(expected, erefs);
+ int prevNumExportCalls = getNumExportCalls();
+ Assert.assertTrue(prevNumExportCalls > 10); // with small buffer there should be lots of exports
+ // calls
+
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ ExportQueue<String, RefUpdates> refExportQueue =
+ ExportQueue.getInstance(RefExporter.QUEUE_ID, fc.getAppConfiguration());
+ try (Transaction tx = fc.newTransaction()) {
+ for (int i = 0; i < 1000; i++) {
+ refExportQueue.add(tx, nk(i), new RefUpdates(ns(i + 12), ns(i + 10)));
+ }
+
+ tx.commit();
+ }
+ }
+
+ miniFluo.waitForObservers();
+
+ erefs = getExportedReferees();
+ expected = new HashMap<>();
+
+ for (int i = 0; i < 1000; i++) {
+ expected.computeIfAbsent(nk(i + 12), s -> new HashSet<>()).add(nk(i));
+ expected.computeIfAbsent(nk(i + 20), s -> new HashSet<>()).add(nk(i));
+ }
+
+ assertEquals(expected, erefs);
+ prevNumExportCalls = getNumExportCalls() - prevNumExportCalls;
+ Assert.assertTrue(prevNumExportCalls > 10);
+ }
+
+ public void assertEquals(Map<String, Set<String>> expected, Map<String, Set<String>> actual) {
+ if (!expected.equals(actual)) {
+ System.out.println("*** diff ***");
+ diff(expected, actual);
+ Assert.fail();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportQueueIT.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportQueueIT.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportQueueIT.java
new file mode 100644
index 0000000..55c1a27
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportQueueIT.java
@@ -0,0 +1,114 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.LoaderExecutor;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ExportQueueIT extends ExportTestBase {
+
+ @Test
+ public void testExport() {
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0999", "0005", "0002"));
+ loader.execute(new DocumentLoader("0002", "0999", "0042"));
+ loader.execute(new DocumentLoader("0005", "0999", "0042"));
+ loader.execute(new DocumentLoader("0042", "0999"));
+ }
+
+ miniFluo.waitForObservers();
+
+ Assert.assertEquals(ns("0002", "0005", "0042"), getExportedReferees("0999"));
+ Assert.assertEquals(ns("0999"), getExportedReferees("0002"));
+ Assert.assertEquals(ns("0999"), getExportedReferees("0005"));
+ Assert.assertEquals(ns("0002", "0005"), getExportedReferees("0042"));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0999", "0005", "0042"));
+ }
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0999", "0005"));
+ }
+
+ miniFluo.waitForObservers();
+
+ Assert.assertEquals(ns("0002", "0005", "0042"), getExportedReferees("0999"));
+ Assert.assertEquals(ns(new String[0]), getExportedReferees("0002"));
+ Assert.assertEquals(ns("0999"), getExportedReferees("0005"));
+ Assert.assertEquals(ns("0002", "0005"), getExportedReferees("0042"));
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0042", "0999", "0002", "0005"));
+ loader.execute(new DocumentLoader("0005", "0002"));
+ }
+
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ loader.execute(new DocumentLoader("0005", "0003"));
+ }
+
+ miniFluo.waitForObservers();
+
+ Assert.assertEquals(ns("0002", "0042"), getExportedReferees("0999"));
+ Assert.assertEquals(ns("0042"), getExportedReferees("0002"));
+ Assert.assertEquals(ns("0005"), getExportedReferees("0003"));
+ Assert.assertEquals(ns("0999", "0042"), getExportedReferees("0005"));
+ Assert.assertEquals(ns("0002"), getExportedReferees("0042"));
+
+ }
+ }
+
+ @Test
+ public void exportStressTest() {
+ FluoConfiguration config = new FluoConfiguration(miniFluo.getClientConfiguration());
+ config.setLoaderQueueSize(100);
+ config.setLoaderThreads(20);
+
+ try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
+
+ loadRandom(fc, 1000, 500);
+
+ miniFluo.waitForObservers();
+
+ diff(getFluoReferees(fc), getExportedReferees());
+
+ assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
+
+ loadRandom(fc, 1000, 500);
+
+ miniFluo.waitForObservers();
+
+ assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
+
+ loadRandom(fc, 1000, 10000);
+
+ miniFluo.waitForObservers();
+
+ assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
+
+ loadRandom(fc, 1000, 10000);
+
+ miniFluo.waitForObservers();
+
+ assertEquals(getFluoReferees(fc), getExportedReferees(), fc);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
new file mode 100644
index 0000000..3c210a9
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/ExportTestBase.java
@@ -0,0 +1,274 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+import com.google.common.collect.Iterators;
+import org.apache.commons.io.FileUtils;
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.FluoFactory;
+import org.apache.fluo.api.client.LoaderExecutor;
+import org.apache.fluo.api.client.Snapshot;
+import org.apache.fluo.api.client.scanner.CellScanner;
+import org.apache.fluo.api.client.scanner.ColumnScanner;
+import org.apache.fluo.api.client.scanner.RowScanner;
+import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ObserverSpecification;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.ColumnValue;
+import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.mini.MiniFluo;
+import org.apache.fluo.recipes.core.export.ExportQueue;
+import org.apache.fluo.recipes.core.export.Exporter;
+import org.apache.fluo.recipes.core.export.SequencedExport;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+
+public class ExportTestBase {
+
+ private static Map<String, Map<String, RefInfo>> globalExports = new HashMap<>();
+ private static int exportCalls = 0;
+
+ protected static Set<String> getExportedReferees(String node) {
+ synchronized (globalExports) {
+ Set<String> ret = new HashSet<>();
+
+ Map<String, RefInfo> referees = globalExports.get(node);
+
+ if (referees == null) {
+ return ret;
+ }
+
+ referees.forEach((k, v) -> {
+ if (!v.deleted)
+ ret.add(k);
+ });
+
+ return ret;
+ }
+ }
+
+ protected static Map<String, Set<String>> getExportedReferees() {
+ synchronized (globalExports) {
+
+ Map<String, Set<String>> ret = new HashMap<>();
+
+ for (String k : globalExports.keySet()) {
+ Set<String> referees = getExportedReferees(k);
+ if (referees.size() > 0) {
+ ret.put(k, referees);
+ }
+ }
+
+ return ret;
+ }
+ }
+
+ protected static int getNumExportCalls() {
+ synchronized (globalExports) {
+ return exportCalls;
+ }
+ }
+
+ public static class RefExporter extends Exporter<String, RefUpdates> {
+
+ public static final String QUEUE_ID = "req";
+
+ private void updateExports(String key, long seq, String addedRef, boolean deleted) {
+ Map<String, RefInfo> referees = globalExports.computeIfAbsent(addedRef, k -> new HashMap<>());
+ referees.compute(key, (k, v) -> (v == null || v.seq < seq) ? new RefInfo(seq, deleted) : v);
+ }
+
+ @Override
+ protected void processExports(Iterator<SequencedExport<String, RefUpdates>> exportIterator) {
+ ArrayList<SequencedExport<String, RefUpdates>> exportList = new ArrayList<>();
+ Iterators.addAll(exportList, exportIterator);
+
+ synchronized (globalExports) {
+ exportCalls++;
+
+ for (SequencedExport<String, RefUpdates> se : exportList) {
+ for (String addedRef : se.getValue().getAddedRefs()) {
+ updateExports(se.getKey(), se.getSequence(), addedRef, false);
+ }
+
+ for (String deletedRef : se.getValue().getDeletedRefs()) {
+ updateExports(se.getKey(), se.getSequence(), deletedRef, true);
+ }
+ }
+ }
+ }
+ }
+
+ protected MiniFluo miniFluo;
+
+ protected int getNumBuckets() {
+ return 13;
+ }
+
+ protected Integer getBufferSize() {
+ return null;
+ }
+
+ @Before
+ public void setUpFluo() throws Exception {
+ FileUtils.deleteQuietly(new File("target/mini"));
+
+ FluoConfiguration props = new FluoConfiguration();
+ props.setApplicationName("eqt");
+ props.setWorkerThreads(20);
+ props.setMiniDataDir("target/mini");
+
+ ObserverSpecification doc = new ObserverSpecification(DocumentObserver.class.getName());
+ props.addObserver(doc);
+
+ SimpleSerializer.setSerializer(props, GsonSerializer.class);
+
+ ExportQueue.Options exportQueueOpts =
+ new ExportQueue.Options(RefExporter.QUEUE_ID, RefExporter.class, String.class,
+ RefUpdates.class, getNumBuckets());
+
+ if (getBufferSize() != null) {
+ exportQueueOpts.setBufferSize(getBufferSize());
+ }
+
+ ExportQueue.configure(props, exportQueueOpts);
+
+ miniFluo = FluoFactory.newMiniFluo(props);
+
+ globalExports.clear();
+ exportCalls = 0;
+ }
+
+ @After
+ public void tearDownFluo() throws Exception {
+ if (miniFluo != null) {
+ miniFluo.close();
+ }
+ }
+
+ protected static Set<String> ns(String... sa) {
+ return new HashSet<>(Arrays.asList(sa));
+ }
+
+ protected static String nk(int i) {
+ return String.format("%06d", i);
+ }
+
+ protected static Set<String> ns(int... ia) {
+ HashSet<String> ret = new HashSet<>();
+ for (int i : ia) {
+ ret.add(nk(i));
+ }
+ return ret;
+ }
+
+ public void assertEquals(Map<String, Set<String>> expected, Map<String, Set<String>> actual,
+ FluoClient fc) {
+ if (!expected.equals(actual)) {
+ System.out.println("*** diff ***");
+ diff(expected, actual);
+ System.out.println("*** fluo dump ***");
+ dump(fc);
+ System.out.println("*** map dump ***");
+
+ Assert.fail();
+ }
+ }
+
+ protected void loadRandom(FluoClient fc, int num, int maxDocId) {
+ try (LoaderExecutor loader = fc.newLoaderExecutor()) {
+ Random rand = new Random();
+
+ for (int i = 0; i < num; i++) {
+ String docid = String.format("%05d", rand.nextInt(maxDocId));
+ String[] refs = new String[rand.nextInt(20) + 1];
+ for (int j = 0; j < refs.length; j++) {
+ refs[j] = String.format("%05d", rand.nextInt(maxDocId));
+ }
+
+ loader.execute(new DocumentLoader(docid, refs));
+ }
+ }
+ }
+
+ protected void diff(Map<String, Set<String>> fr, Map<String, Set<String>> er) {
+ HashSet<String> allKeys = new HashSet<>(fr.keySet());
+ allKeys.addAll(er.keySet());
+
+ for (String k : allKeys) {
+ Set<String> s1 = fr.getOrDefault(k, Collections.emptySet());
+ Set<String> s2 = er.getOrDefault(k, Collections.emptySet());
+
+ HashSet<String> sub1 = new HashSet<>(s1);
+ sub1.removeAll(s2);
+
+ HashSet<String> sub2 = new HashSet<>(s2);
+ sub2.removeAll(s1);
+
+ if (sub1.size() > 0 || sub2.size() > 0) {
+ System.out.println(k + " " + sub1 + " " + sub2);
+ }
+
+ }
+ }
+
+ protected Map<String, Set<String>> getFluoReferees(FluoClient fc) {
+ Map<String, Set<String>> fluoReferees = new HashMap<>();
+
+ try (Snapshot snap = fc.newSnapshot()) {
+
+ Column currCol = new Column("content", "current");
+ RowScanner rowScanner = snap.scanner().over(Span.prefix("d:")).fetch(currCol).byRow().build();
+
+ for (ColumnScanner columnScanner : rowScanner) {
+ String docid = columnScanner.getsRow().substring(2);
+
+ for (ColumnValue columnValue : columnScanner) {
+ String[] refs = columnValue.getsValue().split(" ");
+
+ for (String ref : refs) {
+ if (ref.isEmpty())
+ continue;
+
+ fluoReferees.computeIfAbsent(ref, k -> new HashSet<>()).add(docid);
+ }
+ }
+ }
+ }
+ return fluoReferees;
+ }
+
+ public static void dump(FluoClient fc) {
+ try (Snapshot snap = fc.newSnapshot()) {
+ CellScanner scanner = snap.scanner().build();
+ scanner.forEach(rcv -> System.out.println("row:[" + rcv.getRow() + "] col:["
+ + rcv.getColumn() + "] val:[" + rcv.getValue() + "]"));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/GsonSerializer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/GsonSerializer.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/GsonSerializer.java
new file mode 100644
index 0000000..aa610f7
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/GsonSerializer.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.fluo.recipes.core.export.it;
+
+import java.nio.charset.StandardCharsets;
+
+import com.google.gson.Gson;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.recipes.core.serialization.SimpleSerializer;
+
+public class GsonSerializer implements SimpleSerializer {
+
+ private Gson gson = new Gson();
+
+ @Override
+ public void init(SimpleConfiguration appConfig) {
+
+ }
+
+ @Override
+ public <T> byte[] serialize(T obj) {
+ return gson.toJson(obj).getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public <T> T deserialize(byte[] serObj, Class<T> clazz) {
+ return gson.fromJson(new String(serObj, StandardCharsets.UTF_8), clazz);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefInfo.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefInfo.java
new file mode 100644
index 0000000..e08104b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefInfo.java
@@ -0,0 +1,26 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+class RefInfo {
+ long seq;
+ boolean deleted;
+
+ public RefInfo(long seq, boolean deleted) {
+ this.seq = seq;
+ this.deleted = deleted;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/10ccb16a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefUpdates.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefUpdates.java b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefUpdates.java
new file mode 100644
index 0000000..cd68049
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/recipes/core/export/it/RefUpdates.java
@@ -0,0 +1,43 @@
+/*
+ * 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.fluo.recipes.core.export.it;
+
+import java.util.Set;
+
+public class RefUpdates {
+ private Set<String> addedRefs;
+ private Set<String> deletedRefs;
+
+ public RefUpdates() {}
+
+ public RefUpdates(Set<String> addedRefs, Set<String> deletedRefs) {
+ this.addedRefs = addedRefs;
+ this.deletedRefs = deletedRefs;
+ }
+
+ public Set<String> getAddedRefs() {
+ return addedRefs;
+ }
+
+ public Set<String> getDeletedRefs() {
+ return deletedRefs;
+ }
+
+ @Override
+ public String toString() {
+ return "added:" + addedRefs + " deleted:" + deletedRefs;
+ }
+}