fluo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject incubator-fluo-recipes git commit: Added standard way to setup per exporter configuration.
Date Thu, 22 Sep 2016 16:26:57 GMT
Repository: incubator-fluo-recipes
Updated Branches:
  refs/heads/master 6f5177363 -> dc765718c


Added standard way to setup per exporter configuration.


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/dc765718
Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/tree/dc765718
Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/diff/dc765718

Branch: refs/heads/master
Commit: dc765718c59104a1643cbaa0ae81d565ae5e4146
Parents: 6f51773
Author: Keith Turner <kturner@apache.org>
Authored: Wed Sep 21 18:02:23 2016 -0400
Committer: Keith Turner <keith@deenlo.com>
Committed: Thu Sep 22 12:24:18 2016 -0400

----------------------------------------------------------------------
 docs/accumulo-export-queue.md                   |  24 +-
 .../accumulo/export/AccumuloExportQueue.java    | 278 -------------------
 .../accumulo/export/AccumuloExporter.java       |  84 +++++-
 .../accumulo/export/AccumuloReplicator.java     |   2 +-
 .../recipes/accumulo/export/AccumuloWriter.java | 169 +++++++++++
 .../accumulo/export/AccumuloExportTest.java     |   2 +-
 .../recipes/core/export/ExportObserver.java     |  19 +-
 .../fluo/recipes/core/export/ExportQueue.java   |  32 +++
 .../fluo/recipes/core/export/Exporter.java      |  13 +-
 .../fluo/recipes/core/export/OptionsTest.java   |  12 +-
 .../recipes/test/export/AccumuloExporterIT.java |  12 +-
 .../test/export/AccumuloReplicatorIT.java       |  13 +-
 .../recipes/test/export/SimpleExporter.java     |   2 +-
 13 files changed, 347 insertions(+), 315 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/docs/accumulo-export-queue.md
----------------------------------------------------------------------
diff --git a/docs/accumulo-export-queue.md b/docs/accumulo-export-queue.md
index dde04fb..33a1c97 100644
--- a/docs/accumulo-export-queue.md
+++ b/docs/accumulo-export-queue.md
@@ -3,8 +3,8 @@
 ## Background
 
 The [Export Queue Recipe][1] provides a generic foundation for building export mechanism
to any
-external data store. The [AccumuloExportQueue] provides an implementation of this recipe
for
-Accumulo. The [AccumuloExportQueue] is located the 'fluo-recipes-accumulo' module and provides
the
+external data store. The [AccumuloExporter] provides an implementation of this recipe for
+Accumulo. The [AccumuloExporter] is located the `fluo-recipes-accumulo` module and provides
the
 following functionality:
 
  * Safely batches writes to Accumulo made by multiple transactions exporting data.
@@ -24,7 +24,7 @@ Exporting to Accumulo is easy. Follow the steps below:
     public class SimpleExporter extends AccumuloExporter<String, String> {
 
       @Override
-      protected Collection<Mutation> processExport(SequencedExport<String, String>
export) {
+      protected Collection<Mutation> translate(SequencedExport<String, String>
export) {
         Mutation m = new Mutation(export.getKey());
         m.put("cf", "cq", export.getSequence(), export.getValue());
         return Collections.singleton(m);
@@ -32,7 +32,7 @@ Exporting to Accumulo is easy. Follow the steps below:
     }
     ```
 
-2. With a `SimpleExporter` created, configure a [AccumuloExportQueue] to use `SimpleExporter`
and
+2. With a `SimpleExporter` created, configure an `ExportQueue` to use `SimpleExporter` and
    give it information on how to connect to Accumulo. 
 
     ```java
@@ -46,10 +46,17 @@ Exporting to Accumulo is easy. Follow the steps below:
     String password =       // Accumulo user password
     String exportTable =    // Name of table to export to
 
-    // Configure accumulo export queue
-    AccumuloExportQueue.configure(fluoConfig, new ExportQueue.Options(EXPORT_QUEUE_ID,
-        SimpleExporter.class.getName(), String.class.getName(), String.class.getName(), numMapBuckets),
-        new AccumuloExportQueue.Options(instance, zookeepers, user, password, exportTable));
+
+    // Create config for export table.
+    AccumuloExporter.Configuration exportTableCfg =
+        new AccumuloExporter.Configuration(instance, zookeepers, user, password, exportTable);
+
+    // Create config for export queue.
+    ExportQueue.Options eqOpts = new ExportQueue.Options(EXPORT_QUEUE_ID, SimpleExporter.class,
+        String.class, String.class, numMapBuckets).setExporterConfiguration(exportTableCfg);
+
+    // Configure export queue.  This will modify fluoConfig.
+    ExportQueue.configure(fluoConfig, qeOpts);
 
     // Initialize Fluo using fluoConfig
     ```
@@ -83,7 +90,6 @@ Exporting to Accumulo is easy. Follow the steps below:
 [AccumuloReplicator] is a specialized [AccumuloExporter] that replicates a Fluo table to
Accumulo.
 
 [1]: export-queue.md
-[AccumuloExportQueue]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportQueue.java
 [AccumuloExporter]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
 [AccumuloReplicator]: ../modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/DifferenceExport.java
 

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportQueue.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportQueue.java
b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportQueue.java
deleted file mode 100644
index fdba5f3..0000000
--- a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportQueue.java
+++ /dev/null
@@ -1,278 +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.accumulo.export;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
-import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.BatchWriterConfig;
-import org.apache.accumulo.core.client.ClientConfiguration;
-import org.apache.accumulo.core.client.Connector;
-import org.apache.accumulo.core.client.MutationsRejectedException;
-import org.apache.accumulo.core.client.TableExistsException;
-import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.client.ZooKeeperInstance;
-import org.apache.accumulo.core.client.security.tokens.PasswordToken;
-import org.apache.accumulo.core.data.Mutation;
-import org.apache.fluo.api.config.FluoConfiguration;
-import org.apache.fluo.api.config.SimpleConfiguration;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumn;
-import org.apache.fluo.recipes.core.export.ExportQueue;
-
-public class AccumuloExportQueue {
-
-  /**
-   * Configures AccumuloExportQueue
-   *
-   * @param fc Fluo configuration
-   * @param eqo Export queue options
-   * @param ao Accumulo export queue options
-   */
-  public static void configure(FluoConfiguration fc, ExportQueue.Options eqo, Options ao)
{
-    ExportQueue.configure(fc, eqo);
-    AccumuloWriter.setConfig(fc.getAppConfiguration(), eqo.getQueueId(), ao);
-  }
-
-  /**
-   * Generates Accumulo mutations by comparing the differences between a RowColumn/Bytes
map that is
-   * generated for old and new data and represents how the data should exist in Accumulo.
When
-   * comparing each row/column/value (RCV) of old and new data, mutations are generated using
the
-   * following rules:
-   * <ul>
-   * <li>If old and new data have the same RCV, nothing is done.
-   * <li>If old and new data have same row/column but different values, an update mutation
is
-   * created for the row/column.
-   * <li>If old data has a row/column that is not in the new data, a delete mutation
is generated.
-   * <li>If new data has a row/column that is not in the old data, an insert mutation
is generated.
-   * <li>Only one mutation is generated per row.
-   * <li>The export sequence number is used for the timestamp in the mutation.
-   * </ul>
-   *
-   * @param oldData Map containing old row/column data
-   * @param newData Map containing new row/column data
-   * @param seq Export sequence number
-   */
-  public static Collection<Mutation> generateMutations(long seq, Map<RowColumn,
Bytes> oldData,
-      Map<RowColumn, Bytes> newData) {
-    Map<Bytes, Mutation> mutationMap = new HashMap<>();
-    for (Map.Entry<RowColumn, Bytes> entry : oldData.entrySet()) {
-      RowColumn rc = entry.getKey();
-      if (!newData.containsKey(rc)) {
-        Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
-        m.putDelete(rc.getColumn().getFamily().toArray(), rc.getColumn().getQualifier().toArray(),
-            seq);
-      }
-    }
-    for (Map.Entry<RowColumn, Bytes> entry : newData.entrySet()) {
-      RowColumn rc = entry.getKey();
-      Column col = rc.getColumn();
-      Bytes newVal = entry.getValue();
-      Bytes oldVal = oldData.get(rc);
-      if (oldVal == null || !oldVal.equals(newVal)) {
-        Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
-        m.put(col.getFamily().toArray(), col.getQualifier().toArray(), seq, newVal.toArray());
-      }
-    }
-    return mutationMap.values();
-  }
-
-  /**
-   * Writes mutations to Accumulo using a shared batch writer
-   *
-   * @since 1.0.0
-   */
-  static class AccumuloWriter {
-
-    private static class Mutations {
-      List<Mutation> mutations;
-      CountDownLatch cdl = new CountDownLatch(1);
-
-      Mutations(Collection<Mutation> mutations) {
-        this.mutations = new ArrayList<>(mutations);
-      }
-    }
-
-    /**
-     * Sets AccumuloWriter config in app configuration
-     */
-    static void setConfig(SimpleConfiguration sc, String id, Options ac) {
-      String prefix = "recipes.accumulo.writer." + id;
-      sc.setProperty(prefix + ".instance", ac.instanceName);
-      sc.setProperty(prefix + ".zookeepers", ac.zookeepers);
-      sc.setProperty(prefix + ".user", ac.user);
-      sc.setProperty(prefix + ".password", ac.password);
-      sc.setProperty(prefix + ".table", ac.table);
-    }
-
-    /**
-     * Gets Accumulo Options from app configuration
-     */
-    static Options getConfig(SimpleConfiguration sc, String id) {
-      String prefix = "recipes.accumulo.writer." + id;
-      String instanceName = sc.getString(prefix + ".instance");
-      String zookeepers = sc.getString(prefix + ".zookeepers");
-      String user = sc.getString(prefix + ".user");
-      String password = sc.getString(prefix + ".password");
-      String table = sc.getString(prefix + ".table");
-      return new Options(instanceName, zookeepers, user, password, table);
-    }
-
-    private static class ExportTask implements Runnable {
-
-      private BatchWriter bw;
-
-      ExportTask(String instanceName, String zookeepers, String user, String password, String
table)
-          throws TableNotFoundException, AccumuloException, AccumuloSecurityException {
-        ZooKeeperInstance zki =
-            new ZooKeeperInstance(new ClientConfiguration().withInstance(instanceName).withZkHosts(
-                zookeepers));
-
-        // TODO need to close batch writer
-        Connector conn = zki.getConnector(user, new PasswordToken(password));
-        try {
-          bw = conn.createBatchWriter(table, new BatchWriterConfig());
-        } catch (TableNotFoundException tnfe) {
-          try {
-            conn.tableOperations().create(table);
-          } catch (TableExistsException e) {
-            // nothing to do
-          }
-
-          bw = conn.createBatchWriter(table, new BatchWriterConfig());
-        }
-      }
-
-      @Override
-      public void run() {
-
-        ArrayList<Mutations> exports = new ArrayList<>();
-
-        while (true) {
-          try {
-            exports.clear();
-
-            // gather export from all threads that have placed an item on the queue
-            exports.add(exportQueue.take());
-            exportQueue.drainTo(exports);
-
-            for (Mutations ml : exports) {
-              bw.addMutations(ml.mutations);
-            }
-
-            bw.flush();
-
-            // notify all threads waiting after flushing
-            for (Mutations ml : exports) {
-              ml.cdl.countDown();
-            }
-
-          } catch (InterruptedException | MutationsRejectedException e) {
-            throw new RuntimeException(e);
-          }
-        }
-      }
-
-    }
-
-    private static LinkedBlockingQueue<Mutations> exportQueue = null;
-
-    private AccumuloWriter(String instanceName, String zookeepers, String user, String password,
-        String table) {
-
-      // TODO: fix this write to static and remove findbugs max rank override in pom.xml
-      exportQueue = new LinkedBlockingQueue<>(10000);
-
-      try {
-        Thread queueProcessingTask =
-            new Thread(new ExportTask(instanceName, zookeepers, user, password, table));
-        queueProcessingTask.setDaemon(true);
-        queueProcessingTask.start();
-      } catch (Exception e) {
-        throw new IllegalStateException(e);
-      }
-    }
-
-    private static Map<String, AccumuloWriter> exporters = new HashMap<>();
-
-
-    static AccumuloWriter getInstance(SimpleConfiguration sc, String id) {
-      return getInstance(getConfig(sc, id));
-    }
-
-    static AccumuloWriter getInstance(Options ac) {
-      return getInstance(ac.instanceName, ac.zookeepers, ac.user, ac.password, ac.table);
-    }
-
-    static synchronized AccumuloWriter getInstance(String instanceName, String zookeepers,
-        String user, String password, String table) {
-
-      String key =
-          instanceName + ":" + zookeepers + ":" + user + ":" + password.hashCode() + ":"
+ table;
-
-      AccumuloWriter ret = exporters.get(key);
-
-      if (ret == null) {
-        ret = new AccumuloWriter(instanceName, zookeepers, user, password, table);
-        exporters.put(key, ret);
-      }
-
-      return ret;
-    }
-
-    void write(Collection<Mutation> mutations) {
-      Mutations work = new Mutations(mutations);
-      exportQueue.add(work);
-      try {
-        work.cdl.await();
-      } catch (InterruptedException e) {
-        throw new RuntimeException(e);
-      }
-    }
-
-  }
-
-  /**
-   * Accumulo export queue options
-   *
-   * @since 1.0.0
-   */
-  public static class Options {
-    String instanceName;
-    String zookeepers;
-    String user;
-    String password;
-    String table;
-
-    public Options(String instanceName, String zookeepers, String user, String password,
-        String table) {
-      this.instanceName = instanceName;
-      this.zookeepers = zookeepers;
-      this.user = user;
-      this.password = password;
-      this.table = table;
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
index 54fc8ca..4cb0730 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloExporter.java
@@ -17,27 +17,50 @@ package org.apache.fluo.recipes.accumulo.export;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Iterator;
+import java.util.Map;
 
 import org.apache.accumulo.core.data.Mutation;
-import org.apache.fluo.api.observer.Observer.Context;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.recipes.core.export.ExportQueue;
 import org.apache.fluo.recipes.core.export.Exporter;
 import org.apache.fluo.recipes.core.export.SequencedExport;
 
 /**
- * An Accumulo-specific {@link Exporter} that writes mutations to Accumulo using a
- * {@link AccumuloExportQueue.AccumuloWriter}
+ * An Accumulo-specific {@link Exporter} that writes mutations to Accumulo. For an overview
of how
+ * to use this, see the project level documentation for exporting to Accumulo.
  *
  * @since 1.0.0
  */
 public abstract class AccumuloExporter<K, V> extends Exporter<K, V> {
 
-  private AccumuloExportQueue.AccumuloWriter accumuloWriter;
+  /**
+   * Use this to configure the Accumulo table where an AccumuloExporter's mutations will
be written.
+   * Create and pass to {@link ExportQueue.Options#setExporterConfiguration(SimpleConfiguration)}
+   *
+   * @since 1.0.0
+   */
+  public static class Configuration extends SimpleConfiguration {
+
+    public Configuration(String instanceName, String zookeepers, String user, String password,
+        String table) {
+      super.setProperty("instanceName", instanceName);
+      super.setProperty("zookeepers", zookeepers);
+      super.setProperty("user", user);
+      super.setProperty("password", password);
+      super.setProperty("table", table);
+    }
+  }
+
+  private AccumuloWriter accumuloWriter;
 
   @Override
-  public void init(String queueId, Context context) throws Exception {
-    accumuloWriter =
-        AccumuloExportQueue.AccumuloWriter.getInstance(context.getAppConfiguration(), queueId);
+  public void init(Exporter.Context context) throws Exception {
+    accumuloWriter = AccumuloWriter.getInstance(context.getExporterConfiguration());
   }
 
   @Override
@@ -47,7 +70,7 @@ public abstract class AccumuloExporter<K, V> extends Exporter<K,
V> {
 
     while (exports.hasNext()) {
       SequencedExport<K, V> export = exports.next();
-      buffer.addAll(processExport(export));
+      buffer.addAll(translate(export));
     }
 
     if (buffer.size() > 0) {
@@ -55,6 +78,49 @@ public abstract class AccumuloExporter<K, V> extends Exporter<K,
V> {
     }
   }
 
-  protected abstract Collection<Mutation> processExport(SequencedExport<K, V>
export);
+  protected abstract Collection<Mutation> translate(SequencedExport<K, V> export);
+
+  /**
+   * Generates Accumulo mutations by comparing the differences between a RowColumn/Bytes
map that is
+   * generated for old and new data and represents how the data should exist in Accumulo.
When
+   * comparing each row/column/value (RCV) of old and new data, mutations are generated using
the
+   * following rules:
+   * <ul>
+   * <li>If old and new data have the same RCV, nothing is done.
+   * <li>If old and new data have same row/column but different values, an update mutation
is
+   * created for the row/column.
+   * <li>If old data has a row/column that is not in the new data, a delete mutation
is generated.
+   * <li>If new data has a row/column that is not in the old data, an insert mutation
is generated.
+   * <li>Only one mutation is generated per row.
+   * <li>The export sequence number is used for the timestamp in the mutation.
+   * </ul>
+   *
+   * @param oldData Map containing old row/column data
+   * @param newData Map containing new row/column data
+   * @param seq Export sequence number
+   */
+  public static Collection<Mutation> generateMutations(long seq, Map<RowColumn,
Bytes> oldData,
+      Map<RowColumn, Bytes> newData) {
+    Map<Bytes, Mutation> mutationMap = new HashMap<>();
+    for (Map.Entry<RowColumn, Bytes> entry : oldData.entrySet()) {
+      RowColumn rc = entry.getKey();
+      if (!newData.containsKey(rc)) {
+        Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
+        m.putDelete(rc.getColumn().getFamily().toArray(), rc.getColumn().getQualifier().toArray(),
+            seq);
+      }
+    }
+    for (Map.Entry<RowColumn, Bytes> entry : newData.entrySet()) {
+      RowColumn rc = entry.getKey();
+      Column col = rc.getColumn();
+      Bytes newVal = entry.getValue();
+      Bytes oldVal = oldData.get(rc);
+      if (oldVal == null || !oldVal.equals(newVal)) {
+        Mutation m = mutationMap.computeIfAbsent(rc.getRow(), r -> new Mutation(r.toArray()));
+        m.put(col.getFamily().toArray(), col.getQualifier().toArray(), seq, newVal.toArray());
+      }
+    }
+    return mutationMap.values();
+  }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
index aaea742..abaecfd 100644
--- a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloReplicator.java
@@ -34,7 +34,7 @@ import org.apache.fluo.recipes.core.transaction.TxLog;
 public class AccumuloReplicator extends AccumuloExporter<String, TxLog> {
 
   @Override
-  protected Collection<Mutation> processExport(SequencedExport<String, TxLog>
export) {
+  protected Collection<Mutation> translate(SequencedExport<String, TxLog> export)
{
     return generateMutations(export.getSequence(), export.getValue());
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
new file mode 100644
index 0000000..9016a2c
--- /dev/null
+++ b/modules/accumulo/src/main/java/org/apache/fluo/recipes/accumulo/export/AccumuloWriter.java
@@ -0,0 +1,169 @@
+/*
+ * 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.accumulo.export;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.ClientConfiguration;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.MutationsRejectedException;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.fluo.api.config.SimpleConfiguration;
+
+/**
+ * Writes mutations to Accumulo using a shared batch writer
+ *
+ * @since 1.0.0
+ */
+class AccumuloWriter {
+
+  private static class Mutations {
+    List<Mutation> mutations;
+    CountDownLatch cdl = new CountDownLatch(1);
+
+    Mutations(Collection<Mutation> mutations) {
+      this.mutations = new ArrayList<>(mutations);
+    }
+  }
+
+  private static class ExportTask implements Runnable {
+
+    private BatchWriter bw;
+
+    ExportTask(String instanceName, String zookeepers, String user, String password, String
table)
+        throws TableNotFoundException, AccumuloException, AccumuloSecurityException {
+      ZooKeeperInstance zki =
+          new ZooKeeperInstance(new ClientConfiguration().withInstance(instanceName).withZkHosts(
+              zookeepers));
+
+      // TODO need to close batch writer
+      Connector conn = zki.getConnector(user, new PasswordToken(password));
+      try {
+        bw = conn.createBatchWriter(table, new BatchWriterConfig());
+      } catch (TableNotFoundException tnfe) {
+        try {
+          conn.tableOperations().create(table);
+        } catch (TableExistsException e) {
+          // nothing to do
+        }
+
+        bw = conn.createBatchWriter(table, new BatchWriterConfig());
+      }
+    }
+
+    @Override
+    public void run() {
+
+      ArrayList<AccumuloWriter.Mutations> exports = new ArrayList<>();
+
+      while (true) {
+        try {
+          exports.clear();
+
+          // gather export from all threads that have placed an item on the queue
+          exports.add(exportQueue.take());
+          exportQueue.drainTo(exports);
+
+          for (AccumuloWriter.Mutations ml : exports) {
+            bw.addMutations(ml.mutations);
+          }
+
+          bw.flush();
+
+          // notify all threads waiting after flushing
+          for (AccumuloWriter.Mutations ml : exports) {
+            ml.cdl.countDown();
+          }
+
+        } catch (InterruptedException | MutationsRejectedException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    }
+
+  }
+
+  private static LinkedBlockingQueue<AccumuloWriter.Mutations> exportQueue = null;
+
+  private AccumuloWriter(String instanceName, String zookeepers, String user, String password,
+      String table) {
+
+    // TODO: fix this write to static and remove findbugs max rank override in pom.xml
+    exportQueue = new LinkedBlockingQueue<>(10000);
+
+    try {
+      Thread queueProcessingTask =
+          new Thread(new ExportTask(instanceName, zookeepers, user, password, table));
+      queueProcessingTask.setDaemon(true);
+      queueProcessingTask.start();
+    } catch (Exception e) {
+      throw new IllegalStateException(e);
+    }
+  }
+
+  private static Map<String, AccumuloWriter> exporters = new HashMap<>();
+
+
+  static AccumuloWriter getInstance(SimpleConfiguration sc) {
+    String instanceName = sc.getString("instanceName");
+    String zookeepers = sc.getString("zookeepers");
+    String user = sc.getString("user");
+    String password = sc.getString("password");
+    String table = sc.getString("table");
+    return getInstance(instanceName, zookeepers, user, password, table);
+  }
+
+  static synchronized AccumuloWriter getInstance(String instanceName, String zookeepers,
+      String user, String password, String table) {
+
+    String key =
+        instanceName + ":" + zookeepers + ":" + user + ":" + password.hashCode() + ":" +
table;
+
+    AccumuloWriter ret = exporters.get(key);
+
+    if (ret == null) {
+      ret = new AccumuloWriter(instanceName, zookeepers, user, password, table);
+      exporters.put(key, ret);
+    }
+
+    return ret;
+  }
+
+  void write(Collection<Mutation> mutations) {
+    AccumuloWriter.Mutations work = new Mutations(mutations);
+    exportQueue.add(work);
+    try {
+      work.cdl.await();
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
----------------------------------------------------------------------
diff --git a/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
b/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
index e4be08a..d0ebaee 100644
--- a/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
+++ b/modules/accumulo/src/test/java/org/apache/fluo/recipes/accumulo/export/AccumuloExportTest.java
@@ -45,7 +45,7 @@ public class AccumuloExportTest {
 
   public static Collection<Mutation> genMutations(String key, long seq, Optional<String>
oldVal,
       Optional<String> newVal) {
-    return AccumuloExportQueue.generateMutations(seq, genData(key, oldVal), genData(key,
newVal));
+    return AccumuloExporter.generateMutations(seq, genData(key, oldVal), genData(key, newVal));
   }
 
   public static Mutation makePut(String key, String val, long seq) {

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
index b3070e0..d086577 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportObserver.java
@@ -20,6 +20,7 @@ import java.util.NoSuchElementException;
 
 import com.google.common.collect.Iterators;
 import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.config.SimpleConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.observer.AbstractObserver;
@@ -98,7 +99,23 @@ public class ExportObserver<K, V> extends AbstractObserver {
 
     memLimit = opts.getBufferSize();
 
-    exporter.init(queueId, context);
+    exporter.init(new Exporter.Context() {
+
+      @Override
+      public String getQueueId() {
+        return queueId;
+      }
+
+      @Override
+      public SimpleConfiguration getExporterConfiguration() {
+        return opts.getExporterConfiguration();
+      }
+
+      @Override
+      public Context getObserverContext() {
+        return context;
+      }
+    });
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
index fcc3a74..150503e 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/ExportQueue.java
@@ -20,6 +20,7 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
 import java.util.regex.Pattern;
 
@@ -187,6 +188,7 @@ public class ExportQueue<K, V> {
     String valueType;
     String exporterType;
     String queueId;
+    SimpleConfiguration exporterConfig;
 
     Options(String queueId, SimpleConfiguration appConfig) {
       this.queueId = queueId;
@@ -198,6 +200,8 @@ public class ExportQueue<K, V> {
       this.bufferSize = appConfig.getLong(PREFIX + queueId + ".bufferSize", DEFAULT_BUFFER_SIZE);
       this.bucketsPerTablet =
           appConfig.getInt(PREFIX + queueId + ".bucketsPerTablet", DEFAULT_BUCKETS_PER_TABLET);
+
+      this.exporterConfig = appConfig.subset(PREFIX + queueId + ".exporterCfg");
     }
 
     public Options(String queueId, String exporterType, String keyType, String valueType,
@@ -262,6 +266,24 @@ public class ExportQueue<K, V> {
       return bucketsPerTablet;
     }
 
+    /**
+     * Sets exporter specific configuration. This configuration will be made available to
an
+     * {@link Exporter} via {@link Exporter.Context#getExporterConfiguration()}.
+     */
+    public Options setExporterConfiguration(SimpleConfiguration config) {
+      Objects.requireNonNull(config);
+      this.exporterConfig = config;
+      return this;
+    }
+
+    public SimpleConfiguration getExporterConfiguration() {
+      if (exporterConfig == null) {
+        return new SimpleConfiguration();
+      }
+
+      return exporterConfig;
+    }
+
     public String getQueueId() {
       return queueId;
     }
@@ -275,9 +297,19 @@ public class ExportQueue<K, V> {
       if (bufferSize != null) {
         appConfig.setProperty(PREFIX + queueId + ".bufferSize", bufferSize);
       }
+
       if (bucketsPerTablet != null) {
         appConfig.setProperty(PREFIX + queueId + ".bucketsPerTablet", bucketsPerTablet);
       }
+
+      if (exporterConfig != null) {
+        Iterator<String> keys = exporterConfig.getKeys();
+        while (keys.hasNext()) {
+          String key = keys.next();
+          appConfig.setProperty(PREFIX + queueId + ".exporterCfg." + key,
+              exporterConfig.getRawString(key));
+        }
+      }
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
index f00cfb0..66aa7e8 100644
--- a/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
+++ b/modules/core/src/main/java/org/apache/fluo/recipes/core/export/Exporter.java
@@ -17,14 +17,23 @@ package org.apache.fluo.recipes.core.export;
 
 import java.util.Iterator;
 
-import org.apache.fluo.api.observer.Observer.Context;
+import org.apache.fluo.api.config.SimpleConfiguration;
+import org.apache.fluo.api.observer.Observer;
 
 /**
  * @since 1.0.0
  */
 public abstract class Exporter<K, V> {
 
-  public void init(String queueId, Context observerContext) throws Exception {}
+  public interface Context {
+    String getQueueId();
+
+    SimpleConfiguration getExporterConfiguration();
+
+    Observer.Context getObserverContext();
+  }
+
+  public void init(Exporter.Context exporterContext) throws Exception {}
 
   /**
    * Must be able to handle same key being exported multiple times and key being exported
out of

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/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 b07caea..60982bd 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
@@ -16,6 +16,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.Options;
 import org.junit.Assert;
 import org.junit.Test;
@@ -25,9 +26,13 @@ public class OptionsTest {
   public void testExportQueueOptions() {
     FluoConfiguration conf = new FluoConfiguration();
 
+    SimpleConfiguration ec1 = new SimpleConfiguration();
+    ec1.setProperty("ep1", "ev1");
+    ec1.setProperty("ep2", 3L);
+
     ExportQueue.configure(conf, new Options("Q1", "ET", "KT", "VT", 100));
     ExportQueue.configure(conf, new Options("Q2", "ET2", "KT2", "VT2", 200).setBucketsPerTablet(20)
-        .setBufferSize(1000000));
+        .setBufferSize(1000000).setExporterConfiguration(ec1));
 
     Options opts1 = new Options("Q1", conf.getAppConfiguration());
 
@@ -47,5 +52,10 @@ public class OptionsTest {
     Assert.assertEquals(opts2.bucketsPerTablet.intValue(), 20);
     Assert.assertEquals(opts2.bufferSize.intValue(), 1000000);
 
+    SimpleConfiguration ec2 = opts2.getExporterConfiguration();
+
+    Assert.assertEquals("ev1", ec2.getString("ep1"));
+    Assert.assertEquals(3, ec2.getInt("ep2"));
+
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
----------------------------------------------------------------------
diff --git a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
index 38985f1..d4ad914 100644
--- a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
+++ b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloExporterIT.java
@@ -30,7 +30,7 @@ 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.mini.MiniFluo;
-import org.apache.fluo.recipes.accumulo.export.AccumuloExportQueue;
+import org.apache.fluo.recipes.accumulo.export.AccumuloExporter;
 import org.apache.fluo.recipes.core.export.ExportQueue;
 import org.apache.fluo.recipes.test.AccumuloExportITBase;
 import org.apache.hadoop.io.Text;
@@ -51,10 +51,12 @@ public class AccumuloExporterIT extends AccumuloExportITBase {
 
     MiniAccumuloCluster miniAccumulo = getMiniAccumuloCluster();
 
-    AccumuloExportQueue.configure(getFluoConfiguration(), new ExportQueue.Options(QUEUE_ID,
-        SimpleExporter.class.getName(), String.class.getName(), String.class.getName(), 5)
-        .setBucketsPerTablet(1), new AccumuloExportQueue.Options(miniAccumulo.getInstanceName(),
-        miniAccumulo.getZooKeepers(), ACCUMULO_USER, ACCUMULO_PASSWORD, exportTable));
+    ExportQueue.configure(
+        getFluoConfiguration(),
+        new ExportQueue.Options(QUEUE_ID, SimpleExporter.class, String.class, String.class,
5)
+            .setBucketsPerTablet(1).setExporterConfiguration(
+                new AccumuloExporter.Configuration(miniAccumulo.getInstanceName(), miniAccumulo
+                    .getZooKeepers(), ACCUMULO_USER, ACCUMULO_PASSWORD, exportTable)));
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
----------------------------------------------------------------------
diff --git a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
index e5b4e1a..635f6e9 100644
--- a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
+++ b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/AccumuloReplicatorIT.java
@@ -28,7 +28,7 @@ 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.mini.MiniFluo;
-import org.apache.fluo.recipes.accumulo.export.AccumuloExportQueue;
+import org.apache.fluo.recipes.accumulo.export.AccumuloExporter;
 import org.apache.fluo.recipes.accumulo.export.AccumuloReplicator;
 import org.apache.fluo.recipes.core.export.ExportQueue;
 import org.apache.fluo.recipes.core.transaction.TxLog;
@@ -55,12 +55,11 @@ public class AccumuloReplicatorIT extends AccumuloExportITBase {
 
     MiniAccumuloCluster miniAccumulo = getMiniAccumuloCluster();
 
-    AccumuloExportQueue.configure(
-        getFluoConfiguration(),
-        new ExportQueue.Options(QUEUE_ID, AccumuloReplicator.class.getName(), String.class
-            .getName(), TxLog.class.getName(), 5),
-        new AccumuloExportQueue.Options(miniAccumulo.getInstanceName(), miniAccumulo
-            .getZooKeepers(), ACCUMULO_USER, ACCUMULO_PASSWORD, exportTable));
+    ExportQueue.configure(getFluoConfiguration(), new ExportQueue.Options(QUEUE_ID,
+        AccumuloReplicator.class.getName(), String.class.getName(), TxLog.class.getName(),
5)
+        .setExporterConfiguration(new AccumuloExporter.Configuration(
+            miniAccumulo.getInstanceName(), miniAccumulo.getZooKeepers(), ACCUMULO_USER,
+            ACCUMULO_PASSWORD, exportTable)));
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/incubator-fluo-recipes/blob/dc765718/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
----------------------------------------------------------------------
diff --git a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
index 5aa5812..14c1aa4 100644
--- a/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
+++ b/modules/test/src/test/java/org/apache/fluo/recipes/test/export/SimpleExporter.java
@@ -25,7 +25,7 @@ import org.apache.fluo.recipes.core.export.SequencedExport;
 public class SimpleExporter extends AccumuloExporter<String, String> {
 
   @Override
-  protected Collection<Mutation> processExport(SequencedExport<String, String>
export) {
+  protected Collection<Mutation> translate(SequencedExport<String, String> export)
{
     Mutation m = new Mutation(export.getKey());
     m.put("cf", "cq", export.getSequence(), export.getValue());
     return Collections.singleton(m);



Mime
View raw message