You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2012/09/18 16:25:40 UTC
svn commit: r1387182 [1/3] - in /accumulo/trunk:
core/src/main/java/org/apache/accumulo/core/client/
core/src/main/java/org/apache/accumulo/core/client/impl/
core/src/main/java/org/apache/accumulo/core/client/mapreduce/
core/src/main/java/org/apache/ac...
Author: kturner
Date: Tue Sep 18 14:25:37 2012
New Revision: 1387182
URL: http://svn.apache.org/viewvc?rev=1387182&view=rev
Log:
ACCUMULO-706 Added timeout to batch writer
Added:
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/BatchWriterConfig.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/SlowConstraint.java
Modified:
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/Connector.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/BatchWriterImpl.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/ConnectorImpl.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/MultiTableBatchWriterImpl.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchDeleter.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mock/MockConnector.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteCommand.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteManyCommand.java
accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/InsertCommand.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/ClientSideIteratorTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/admin/FindMaxTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloRowInputFormatTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/IntersectingIteratorTest.java
accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/RowFilterTest.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RandomBatchWriter.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/ReadWriteExample.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RowOperations.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/SequentialBatchWriter.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/FileCount.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/Ingest.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataIngest.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/helloworld/InsertWithBatchWriter.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/isolation/InterferenceTest.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Index.java
accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Reverse.java
accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/dirlist/CountTest.java
accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputFormatTest.java
accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputStreamTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/gc/SimpleGarbageCollector.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/Master.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/state/MetaDataStateStore.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/ImportTable.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/master/tableOps/TableRangeOp.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/CreateTestTable.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/GCLotsOfCandidatesTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/TestBinaryRows.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/TestIngest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/TestMultiTableIngest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/TestRandomDeletes.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/continuous/ContinuousIngest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/AddSplitTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BadIteratorMincTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BatchScanSplitTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BatchWriterFlushTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/BloomFilterTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/ConcurrencyTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/ConstraintTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/CreateAndUseTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/DeleteEverythingTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/DeleteRowsSplitTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/DeleteRowsTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/LargeRowTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/LogicalTimeTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/MergeTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/PermissionsTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/RowDeleteTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/ScanIteratorTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/ScanRangeTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/ScanSessionTimeOutTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/ServerSideErrorTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/SparseColumnFamilyTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/TimeoutTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/functional/VisibilityTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/performance/metadata/MetadataBatchScanTest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/randomwalk/State.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/BatchWrite.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/randomwalk/security/TableOp.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/randomwalk/shard/Reindex.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/test/scalability/Ingest.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/trace/TraceServer.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/AddFilesWithMissingEntries.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/MetadataTable.java
accumulo/trunk/server/src/main/java/org/apache/accumulo/server/util/RandomWriter.java
accumulo/trunk/server/src/test/java/org/apache/accumulo/server/gc/TestConfirmDeletes.java
accumulo/trunk/server/src/test/java/org/apache/accumulo/server/master/TestMergeState.java
accumulo/trunk/server/src/test/java/org/apache/accumulo/server/test/iterator/RegExTest.java
accumulo/trunk/server/src/test/java/org/apache/accumulo/server/util/CloneTest.java
accumulo/trunk/server/src/test/java/org/apache/accumulo/server/util/TabletIteratorTest.java
Added: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/BatchWriterConfig.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/BatchWriterConfig.java?rev=1387182&view=auto
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/BatchWriterConfig.java (added)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/BatchWriterConfig.java Tue Sep 18 14:25:37 2012
@@ -0,0 +1,108 @@
+/**
+ * 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.accumulo.core.client;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This object holds configuration settings used to instantiate a {@link BatchWriter}
+ */
+public class BatchWriterConfig {
+ private long maxMemory = 50 * 1024 * 1024;
+ private long maxLatency = Long.MAX_VALUE;
+ private long timeout = Long.MAX_VALUE;
+ private int maxWriteThreads = 3;
+
+ /**
+ *
+ * @param maxMemory
+ * size in bytes of the maximum memory to batch before writing. Defaults to 50M.
+ * @return
+ */
+
+ public BatchWriterConfig setMaxMemory(long maxMemory) {
+ this.maxMemory = maxMemory;
+ return this;
+ }
+
+ /**
+ * @param maxLatency
+ * The maximum amount of time to hold data in memory before flushing it to servers. For no max set to zero or Long.MAX_VALUE with TimeUnit.MILLIS.
+ * Defaults to no max.
+ * @param timeUnit
+ * Determines how maxLatency will be interpreted.
+ * @return this to allow chaining of set methods
+ */
+
+ public BatchWriterConfig setMaxLatency(long maxLatency, TimeUnit timeUnit) {
+ if (maxLatency < 0)
+ throw new IllegalArgumentException("Negative max latency not allowed " + maxLatency);
+
+ this.maxLatency = timeUnit.toMillis(maxLatency);
+ return this;
+ }
+
+ /**
+ *
+ * @param timeout
+ * The maximum amount of time an unresponsive server will be retried. When this timeout is exceeded, the BatchWriter should throw an exception. For
+ * no timeout set to zero or Long.MAX_VALUE with TimeUnit.MILLIS. Defaults to no timeout.
+ * @param timeUnit
+ * @return this to allow chaining of set methods
+ */
+
+ public BatchWriterConfig setTimeout(long timeout, TimeUnit timeUnit) {
+ if (timeout < 0)
+ throw new IllegalArgumentException("Negative timeout not allowed " + timeout);
+
+ if (timeout == 0)
+ timeout = Long.MAX_VALUE;
+ else
+ this.timeout = timeUnit.toMillis(timeout);
+ return this;
+ }
+
+ /**
+ * @param maxWriteThreads
+ * the maximum number of threads to use for writing data to the tablet servers. Defaults to 3.
+ * @return this to allow chaining of set methods
+ */
+
+ public BatchWriterConfig setMaxWriteThreads(int maxWriteThreads) {
+ if (maxWriteThreads <= 0)
+ throw new IllegalArgumentException("Max threads must be positive " + maxWriteThreads);
+
+ this.maxWriteThreads = maxWriteThreads;
+ return this;
+ }
+
+ public long getMaxMemory() {
+ return maxMemory;
+ }
+
+ public long getMaxLatency(TimeUnit timeUnit) {
+ return timeUnit.convert(maxLatency, TimeUnit.MILLISECONDS);
+ }
+
+ public long getTimeout(TimeUnit timeUnit) {
+ return timeUnit.convert(timeout, TimeUnit.MILLISECONDS);
+ }
+
+ public int getMaxWriteThreads() {
+ return maxWriteThreads;
+ }
+}
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/Connector.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/Connector.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/Connector.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/Connector.java Tue Sep 18 14:25:37 2012
@@ -99,13 +99,36 @@ public class Connector {
* @return BatchDeleter object for configuring and deleting
* @throws TableNotFoundException
* when the specified table doesn't exist
+ * @deprecated As of 1.5, replaced by {@link #createBatchDeleter(String, Authorizations, int, BatchWriterConfig)}
*/
+ @Deprecated
public BatchDeleter createBatchDeleter(String tableName, Authorizations authorizations, int numQueryThreads, long maxMemory, long maxLatency,
int maxWriteThreads) throws TableNotFoundException {
return impl.createBatchDeleter(tableName, authorizations, numQueryThreads, maxMemory, maxLatency, maxWriteThreads);
}
/**
+ *
+ * @param tableName
+ * the name of the table to query and delete from
+ * @param authorizations
+ * A set of authorization labels that will be checked against the column visibility of each key in order to filter data. The authorizations passed in
+ * must be a subset of the accumulo user's set of authorizations. If the accumulo user has authorizations (A1, A2) and authorizations (A2, A3) are
+ * passed, then an exception will be thrown.
+ * @param numQueryThreads
+ * the number of concurrent threads to spawn for querying
+ * @param config
+ * configuration used to create batch writer
+ * @return BatchDeleter object for configuring and deleting
+ * @throws TableNotFoundException
+ */
+
+ public BatchDeleter createBatchDeleter(String tableName, Authorizations authorizations, int numQueryThreads, BatchWriterConfig config)
+ throws TableNotFoundException {
+ return impl.createBatchDeleter(tableName, authorizations, numQueryThreads, config);
+ }
+
+ /**
* Factory method to create a BatchWriter connected to Accumulo.
*
* @param tableName
@@ -120,12 +143,29 @@ public class Connector {
* @return BatchWriter object for configuring and writing data to
* @throws TableNotFoundException
* when the specified table doesn't exist
+ * @deprecated As of 1.5, replaced by {@link #createBatchWriter(String, BatchWriterConfig)}
*/
+ @Deprecated
public BatchWriter createBatchWriter(String tableName, long maxMemory, long maxLatency, int maxWriteThreads) throws TableNotFoundException {
return impl.createBatchWriter(tableName, maxMemory, maxLatency, maxWriteThreads);
}
/**
+ * Factory method to create a BatchWriter connected to Accumulo.
+ *
+ * @param tableName
+ * the name of the table to insert data into
+ * @param config
+ * configuration used to create batch writer
+ * @return BatchWriter object for configuring and writing data to
+ * @throws TableNotFoundException
+ */
+
+ public BatchWriter createBatchWriter(String tableName, BatchWriterConfig config) throws TableNotFoundException {
+ return impl.createBatchWriter(tableName, config);
+ }
+
+ /**
* Factory method to create a Multi-Table BatchWriter connected to Accumulo. Multi-table batch writers can queue data for multiple tables, which is good for
* ingesting data into multiple tables from the same source
*
@@ -137,12 +177,27 @@ public class Connector {
* the maximum number of threads to use for writing data to the tablet servers
*
* @return MultiTableBatchWriter object for configuring and writing data to
+ * @deprecated As of 1.5, replaced by {@link #createMultiTableBatchWriter(BatchWriterConfig)}
*/
+ @Deprecated
public MultiTableBatchWriter createMultiTableBatchWriter(long maxMemory, long maxLatency, int maxWriteThreads) {
return impl.createMultiTableBatchWriter(maxMemory, maxLatency, maxWriteThreads);
}
/**
+ * Factory method to create a Multi-Table BatchWriter connected to Accumulo. Multi-table batch writers can queue data for multiple tables. Also data for
+ * multiple tables can be sent to a server in a single batch. Its an efficient way to ingest data into multiple tables from a single process.
+ *
+ * @param config
+ * configuration used to create multi-table batch writer
+ * @return MultiTableBatchWriter object for configuring and writing data to
+ */
+
+ public MultiTableBatchWriter createMultiTableBatchWriter(BatchWriterConfig config) {
+ return impl.createMultiTableBatchWriter(config);
+ }
+
+ /**
* Factory method to create a Scanner connected to Accumulo.
*
* @param tableName
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/BatchWriterImpl.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/BatchWriterImpl.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/BatchWriterImpl.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/BatchWriterImpl.java Tue Sep 18 14:25:37 2012
@@ -17,6 +17,7 @@
package org.apache.accumulo.core.client.impl;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.data.Mutation;
@@ -28,10 +29,10 @@ public class BatchWriterImpl implements
private String table;
private TabletServerBatchWriter bw;
- public BatchWriterImpl(Instance instance, AuthInfo credentials, String table, long maxMemory, long maxLatency, int maxWriteThreads) {
+ public BatchWriterImpl(Instance instance, AuthInfo credentials, String table, BatchWriterConfig config) {
ArgumentChecker.notNull(instance, credentials, table);
this.table = table;
- this.bw = new TabletServerBatchWriter(instance, credentials, maxMemory, maxLatency, maxWriteThreads);
+ this.bw = new TabletServerBatchWriter(instance, credentials, config);
}
@Override
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/ConnectorImpl.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/ConnectorImpl.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/ConnectorImpl.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/ConnectorImpl.java Tue Sep 18 14:25:37 2012
@@ -17,6 +17,7 @@
package org.apache.accumulo.core.client.impl;
import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
import org.apache.accumulo.cloudtrace.instrument.Tracer;
import org.apache.accumulo.core.client.AccumuloException;
@@ -24,6 +25,7 @@ import org.apache.accumulo.core.client.A
import org.apache.accumulo.core.client.BatchDeleter;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
@@ -106,18 +108,34 @@ public class ConnectorImpl extends Conne
public BatchDeleter createBatchDeleter(String tableName, Authorizations authorizations, int numQueryThreads, long maxMemory, long maxLatency,
int maxWriteThreads) throws TableNotFoundException {
ArgumentChecker.notNull(tableName, authorizations);
- return new TabletServerBatchDeleter(instance, credentials, getTableId(tableName), authorizations, numQueryThreads, maxMemory, maxLatency, maxWriteThreads);
+ return new TabletServerBatchDeleter(instance, credentials, getTableId(tableName), authorizations, numQueryThreads, new BatchWriterConfig()
+ .setMaxMemory(maxMemory).setMaxLatency(maxLatency, TimeUnit.MILLISECONDS).setMaxWriteThreads(maxWriteThreads));
+ }
+
+ @Override
+ public BatchDeleter createBatchDeleter(String tableName, Authorizations authorizations, int numQueryThreads, BatchWriterConfig config)
+ throws TableNotFoundException {
+ ArgumentChecker.notNull(tableName, authorizations);
+ return new TabletServerBatchDeleter(instance, credentials, getTableId(tableName), authorizations, numQueryThreads, config);
}
@Override
public BatchWriter createBatchWriter(String tableName, long maxMemory, long maxLatency, int maxWriteThreads) throws TableNotFoundException {
ArgumentChecker.notNull(tableName);
- return new BatchWriterImpl(instance, credentials, getTableId(tableName), maxMemory, maxLatency, maxWriteThreads);
+ return new BatchWriterImpl(instance, credentials, getTableId(tableName), new BatchWriterConfig().setMaxMemory(maxMemory)
+ .setMaxLatency(maxLatency, TimeUnit.MILLISECONDS).setMaxWriteThreads(maxWriteThreads));
+ }
+
+ @Override
+ public BatchWriter createBatchWriter(String tableName, BatchWriterConfig config) throws TableNotFoundException {
+ ArgumentChecker.notNull(tableName);
+ return new BatchWriterImpl(instance, credentials, getTableId(tableName), config);
}
@Override
public MultiTableBatchWriter createMultiTableBatchWriter(long maxMemory, long maxLatency, int maxWriteThreads) {
- return new MultiTableBatchWriterImpl(instance, credentials, maxMemory, maxLatency, maxWriteThreads);
+ return new MultiTableBatchWriterImpl(instance, credentials, new BatchWriterConfig().setMaxMemory(maxMemory)
+ .setMaxLatency(maxLatency, TimeUnit.MILLISECONDS).setMaxWriteThreads(maxWriteThreads));
}
@Override
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/MultiTableBatchWriterImpl.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/MultiTableBatchWriterImpl.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/MultiTableBatchWriterImpl.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/MultiTableBatchWriterImpl.java Tue Sep 18 14:25:37 2012
@@ -21,6 +21,7 @@ import java.util.HashMap;
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.Instance;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
import org.apache.accumulo.core.client.MutationsRejectedException;
@@ -71,10 +72,10 @@ public class MultiTableBatchWriterImpl i
private HashMap<String,BatchWriter> tableWriters;
private Instance instance;
- public MultiTableBatchWriterImpl(Instance instance, AuthInfo credentials, long maxMemory, long maxLatency, int maxWriteThreads) {
+ public MultiTableBatchWriterImpl(Instance instance, AuthInfo credentials, BatchWriterConfig config) {
ArgumentChecker.notNull(instance, credentials);
this.instance = instance;
- this.bw = new TabletServerBatchWriter(instance, credentials, maxMemory, maxLatency, maxWriteThreads);
+ this.bw = new TabletServerBatchWriter(instance, credentials, config);
tableWriters = new HashMap<String,BatchWriter>();
this.closed = false;
}
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchDeleter.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchDeleter.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchDeleter.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchDeleter.java Tue Sep 18 14:25:37 2012
@@ -21,6 +21,7 @@ import java.util.Map.Entry;
import org.apache.accumulo.core.client.BatchDeleter;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
@@ -38,19 +39,15 @@ public class TabletServerBatchDeleter ex
private Instance instance;
private AuthInfo credentials;
private String tableId;
- private long maxMemory;
- private long maxLatency;
- private int maxWriteThreads;
+ private BatchWriterConfig bwConfig;
- public TabletServerBatchDeleter(Instance instance, AuthInfo credentials, String tableId, Authorizations authorizations, int numQueryThreads, long maxMemory,
- long maxLatency, int maxWriteThreads) throws TableNotFoundException {
+ public TabletServerBatchDeleter(Instance instance, AuthInfo credentials, String tableId, Authorizations authorizations, int numQueryThreads,
+ BatchWriterConfig bwConfig) throws TableNotFoundException {
super(instance, credentials, tableId, authorizations, numQueryThreads);
this.instance = instance;
this.credentials = credentials;
this.tableId = tableId;
- this.maxMemory = maxMemory;
- this.maxLatency = maxLatency;
- this.maxWriteThreads = maxWriteThreads;
+ this.bwConfig = bwConfig;
super.addScanIterator(new IteratorSetting(Integer.MAX_VALUE, BatchDeleter.class.getName() + ".NOVALUE", SortedKeyIterator.class));
}
@@ -58,7 +55,7 @@ public class TabletServerBatchDeleter ex
public void delete() throws MutationsRejectedException, TableNotFoundException {
BatchWriter bw = null;
try {
- bw = new BatchWriterImpl(instance, credentials, tableId, maxMemory, maxLatency, maxWriteThreads);
+ bw = new BatchWriterImpl(instance, credentials, tableId, bwConfig);
Iterator<Entry<Key,Value>> iter = super.iterator();
while (iter.hasNext()) {
Entry<Key,Value> next = iter.next();
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java Tue Sep 18 14:25:37 2012
@@ -34,6 +34,7 @@ import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.accumulo.cloudtrace.instrument.Span;
@@ -43,11 +44,13 @@ import org.apache.accumulo.cloudtrace.th
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
+import org.apache.accumulo.core.client.TimedOutException;
import org.apache.accumulo.core.client.impl.TabletLocator.TabletServerMutations;
import org.apache.accumulo.core.constraints.Violations;
import org.apache.accumulo.core.data.ConstraintViolationSummary;
@@ -120,6 +123,8 @@ public class TabletServerBatchWriter {
private long maxLatency;
+ private long timeout;
+
private long lastProcessingStartTime;
private long totalAdded = 0;
@@ -142,11 +147,48 @@ public class TabletServerBatchWriter {
private Throwable lastUnknownError = null;
- public TabletServerBatchWriter(Instance instance, AuthInfo credentials, long maxMemory, long maxLatency, int numSendThreads) {
+ private Map<String,TimeoutTracker> timeoutTrackers;
+
+ private static class TimeoutTracker {
+
+ String server;
+ long timeOut;
+ long activityTime;
+ Long firstErrorTime = null;
+
+ TimeoutTracker(String server, long timeOut) {
+ this.timeOut = timeOut;
+ this.server = server;
+ }
+
+ void startingWrite() {
+ activityTime = System.currentTimeMillis();
+ }
+
+ void madeProgress() {
+ activityTime = System.currentTimeMillis();
+ firstErrorTime = null;
+ }
+
+ void errorOccured(Exception e) {
+ if (firstErrorTime == null) {
+ firstErrorTime = activityTime;
+ } else if (System.currentTimeMillis() - firstErrorTime > timeOut) {
+ throw new TimedOutException(Collections.singleton(server));
+ }
+ }
+
+ public long getTimeOut() {
+ return timeOut;
+ }
+ }
+
+ public TabletServerBatchWriter(Instance instance, AuthInfo credentials, BatchWriterConfig config) {
this.instance = instance;
- this.maxMem = maxMemory;
- this.maxLatency = maxLatency <= 0 ? Long.MAX_VALUE : maxLatency;
+ this.maxMem = config.getMaxMemory();
+ this.maxLatency = config.getMaxLatency(TimeUnit.MILLISECONDS) <= 0 ? Long.MAX_VALUE : config.getMaxLatency(TimeUnit.MILLISECONDS);
this.credentials = credentials;
+ this.timeout = config.getTimeout(TimeUnit.MILLISECONDS);
mutations = new MutationSet();
violations = new Violations();
@@ -158,9 +200,11 @@ public class TabletServerBatchWriter {
jtimer = new Timer("BatchWriterLatencyTimer", true);
- writer = new MutationWriter(numSendThreads);
+ writer = new MutationWriter(config.getMaxWriteThreads());
failedMutations = new FailedMutations();
+ timeoutTrackers = Collections.synchronizedMap(new HashMap<String,TabletServerBatchWriter.TimeoutTracker>());
+
if (this.maxLatency != Long.MAX_VALUE) {
jtimer.schedule(new TimerTask() {
public void run() {
@@ -446,7 +490,7 @@ public class TabletServerBatchWriter {
unknownErrors++;
this.lastUnknownError = t;
this.notifyAll();
- if (t instanceof TableDeletedException || t instanceof TableOfflineException)
+ if (t instanceof TableDeletedException || t instanceof TableOfflineException || t instanceof TimedOutException)
log.debug(msg, t); // this is not unknown
else
log.error(msg, t);
@@ -700,8 +744,15 @@ public class TabletServerBatchWriter {
Span span = Trace.start("sendMutations");
try {
+
+ TimeoutTracker timeoutTracker = timeoutTrackers.get(location);
+ if (timeoutTracker == null) {
+ timeoutTracker = new TimeoutTracker(location, timeout);
+ timeoutTrackers.put(location, timeoutTracker);
+ }
+
long st1 = System.currentTimeMillis();
- failures = sendMutationsToTabletServer(location, mutationBatch);
+ failures = sendMutationsToTabletServer(location, mutationBatch, timeoutTracker);
long st2 = System.currentTimeMillis();
if (log.isTraceEnabled())
log.trace("sent " + String.format("%,d", count) + " mutations to " + location + " in "
@@ -743,7 +794,8 @@ public class TabletServerBatchWriter {
}
}
- private MutationSet sendMutationsToTabletServer(String location, Map<KeyExtent,List<Mutation>> tabMuts) throws IOException, AccumuloSecurityException,
+ private MutationSet sendMutationsToTabletServer(String location, Map<KeyExtent,List<Mutation>> tabMuts, TimeoutTracker timeoutTracker) throws IOException,
+ AccumuloSecurityException,
AccumuloServerException {
if (tabMuts.size() == 0) {
return new MutationSet();
@@ -751,8 +803,15 @@ public class TabletServerBatchWriter {
TInfo tinfo = Tracer.traceInfo();
TTransport transport = null;
+ timeoutTracker.startingWrite();
+
try {
- TabletClientService.Iface client = ThriftUtil.getTServerClient(location, instance.getConfiguration());
+ TabletClientService.Iface client;
+ if (timeoutTracker.getTimeOut() < Long.MAX_VALUE)
+ client = ThriftUtil.getTServerClient(location, instance.getConfiguration(), timeoutTracker.getTimeOut());
+ else
+ client = ThriftUtil.getTServerClient(location, instance.getConfiguration());
+
try {
MutationSet allFailures = new MutationSet();
@@ -767,9 +826,11 @@ public class TabletServerBatchWriter {
} catch (ConstraintViolationException e) {
updatedConstraintViolations(Translator.translate(e.violationSummaries, Translator.TCVST));
}
+ timeoutTracker.madeProgress();
} else {
long usid = client.startUpdate(tinfo, credentials);
+ timeoutTracker.madeProgress();
List<TMutation> updates = new ArrayList<TMutation>();
for (Entry<KeyExtent,List<Mutation>> entry : tabMuts.entrySet()) {
@@ -783,12 +844,15 @@ public class TabletServerBatchWriter {
}
client.applyUpdates(tinfo, usid, entry.getKey().toThrift(), updates);
+ timeoutTracker.madeProgress();
updates.clear();
size = 0;
}
}
UpdateErrors updateErrors = client.closeUpdate(tinfo, usid);
+ timeoutTracker.madeProgress();
+
Map<KeyExtent,Long> failures = Translator.translate(updateErrors.failedExtents, Translator.TKET);
updatedConstraintViolations(Translator.translate(updateErrors.violationSummaries, Translator.TCVST));
updateAuthorizationFailures(Translator.translate(updateErrors.authorizationFailures, Translator.TKET));
@@ -810,6 +874,7 @@ public class TabletServerBatchWriter {
ThriftUtil.returnClient((TServiceClient) client);
}
} catch (TTransportException e) {
+ timeoutTracker.errorOccured(e);
throw new IOException(e);
} catch (TApplicationException tae) {
updateServerErrors(location, tae);
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java Tue Sep 18 14:25:37 2012
@@ -19,10 +19,12 @@ package org.apache.accumulo.core.client.
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.concurrent.TimeUnit;
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.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
@@ -90,6 +92,7 @@ public class AccumuloOutputFormat extend
private static final String MAX_MUTATION_BUFFER_SIZE = PREFIX + ".maxmemory";
private static final String MAX_LATENCY = PREFIX + ".maxlatency";
private static final String NUM_WRITE_THREADS = PREFIX + ".writethreads";
+ private static final String TIMEOUT = PREFIX + ".timeout";
private static final long DEFAULT_MAX_MUTATION_BUFFER_SIZE = 50 * 1024 * 1024; // 50MB
private static final int DEFAULT_MAX_LATENCY = 60 * 1000; // 1 minute
@@ -156,18 +159,38 @@ public class AccumuloOutputFormat extend
conf.set(INSTANCE_NAME, instanceName);
}
+ /**
+ * see {@link BatchWriterConfig#setMaxMemory(long)}
+ */
+
public static void setMaxMutationBufferSize(Configuration conf, long numberOfBytes) {
conf.setLong(MAX_MUTATION_BUFFER_SIZE, numberOfBytes);
}
+ /**
+ * see {@link BatchWriterConfig#setMaxLatency(long, TimeUnit)}
+ */
+
public static void setMaxLatency(Configuration conf, int numberOfMilliseconds) {
conf.setInt(MAX_LATENCY, numberOfMilliseconds);
}
+ /**
+ * see {@link BatchWriterConfig#setMaxWriteThreads(int)}
+ */
+
public static void setMaxWriteThreads(Configuration conf, int numberOfThreads) {
conf.setInt(NUM_WRITE_THREADS, numberOfThreads);
}
+ /**
+ * see {@link BatchWriterConfig#setTimeout(long, TimeUnit)}
+ */
+
+ public static void setTimeout(Configuration conf, long time, TimeUnit timeUnit) {
+ conf.setLong(TIMEOUT, timeUnit.toMillis(time));
+ }
+
public static void setLogLevel(Configuration conf, Level level) {
ArgumentChecker.notNull(level);
conf.setInt(LOGLEVEL, level.toInt());
@@ -223,6 +246,10 @@ public class AccumuloOutputFormat extend
return conf.getInt(NUM_WRITE_THREADS, DEFAULT_NUM_WRITE_THREADS);
}
+ protected static long getTimeout(Configuration conf) {
+ return conf.getLong(TIMEOUT, Long.MAX_VALUE);
+ }
+
protected static Level getLogLevel(Configuration conf) {
if (conf.get(LOGLEVEL) != null)
return Level.toLevel(conf.getInt(LOGLEVEL, Level.INFO.toInt()));
@@ -263,7 +290,9 @@ public class AccumuloOutputFormat extend
if (!simulate) {
this.conn = getInstance(conf).getConnector(getUsername(conf), getPassword(conf));
- mtbw = conn.createMultiTableBatchWriter(getMaxMutationBufferSize(conf), getMaxLatency(conf), getMaxWriteThreads(conf));
+ mtbw = conn.createMultiTableBatchWriter(new BatchWriterConfig().setMaxMemory(getMaxMutationBufferSize(conf))
+ .setMaxLatency(getMaxLatency(conf), TimeUnit.MILLISECONDS).setMaxWriteThreads(getMaxWriteThreads(conf))
+ .setTimeout(getTimeout(conf), TimeUnit.MILLISECONDS));
}
}
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mock/MockConnector.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mock/MockConnector.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mock/MockConnector.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/client/mock/MockConnector.java Tue Sep 18 14:25:37 2012
@@ -16,9 +16,12 @@
*/
package org.apache.accumulo.core.client.mock;
+import java.util.concurrent.TimeUnit;
+
import org.apache.accumulo.core.client.BatchDeleter;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
@@ -61,6 +64,13 @@ public class MockConnector extends Conne
}
@Override
+ public BatchDeleter createBatchDeleter(String tableName, Authorizations authorizations, int numQueryThreads, BatchWriterConfig config)
+ throws TableNotFoundException {
+ return createBatchDeleter(tableName, authorizations, numQueryThreads, config.getMaxMemory(), config.getMaxLatency(TimeUnit.MILLISECONDS),
+ config.getMaxWriteThreads());
+ }
+
+ @Override
public BatchWriter createBatchWriter(String tableName, long maxMemory, long maxLatency, int maxWriteThreads) throws TableNotFoundException {
if (acu.tables.get(tableName) == null)
throw new TableNotFoundException(tableName, tableName, "no such table");
@@ -68,11 +78,21 @@ public class MockConnector extends Conne
}
@Override
+ public BatchWriter createBatchWriter(String tableName, BatchWriterConfig config) throws TableNotFoundException {
+ return createBatchWriter(tableName, config.getMaxMemory(), config.getMaxLatency(TimeUnit.MILLISECONDS), config.getMaxWriteThreads());
+ }
+
+ @Override
public MultiTableBatchWriter createMultiTableBatchWriter(long maxMemory, long maxLatency, int maxWriteThreads) {
return new MockMultiTableBatchWriter(acu);
}
@Override
+ public MultiTableBatchWriter createMultiTableBatchWriter(BatchWriterConfig config) {
+ return createMultiTableBatchWriter(config.getMaxMemory(), config.getMaxLatency(TimeUnit.MILLISECONDS), config.getMaxWriteThreads());
+ }
+
+ @Override
public Scanner createScanner(String tableName, Authorizations authorizations) throws TableNotFoundException {
MockTable table = acu.tables.get(tableName);
if (table == null)
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteCommand.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteCommand.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteCommand.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteCommand.java Tue Sep 18 14:25:37 2012
@@ -21,6 +21,7 @@ import java.io.IOException;
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.TableNotFoundException;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.security.ColumnVisibility;
@@ -55,7 +56,8 @@ public class DeleteCommand extends Comma
} else {
m.putDelete(colf, colq);
}
- final BatchWriter bw = shellState.getConnector().createBatchWriter(shellState.getTableName(), m.estimatedMemoryUsed() + 0L, 0L, 1);
+ final BatchWriter bw = shellState.getConnector().createBatchWriter(shellState.getTableName(),
+ new BatchWriterConfig().setMaxMemory(m.estimatedMemoryUsed()).setMaxWriteThreads(1));
bw.addMutation(m);
bw.close();
return 0;
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteManyCommand.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteManyCommand.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteManyCommand.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/DeleteManyCommand.java Tue Sep 18 14:25:37 2012
@@ -17,6 +17,7 @@
package org.apache.accumulo.core.util.shell.commands;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.iterators.SortedKeyIterator;
@@ -50,7 +51,7 @@ public class DeleteManyCommand extends S
fetchColumns(cl, scanner, interpeter);
// output / delete the records
- final BatchWriter writer = shellState.getConnector().createBatchWriter(tableName, 1024 * 1024, 1000L, 4);
+ final BatchWriter writer = shellState.getConnector().createBatchWriter(tableName, new BatchWriterConfig());
shellState.printLines(new DeleterFormatter(writer, scanner, cl.hasOption(timestampOpt.getOpt()), shellState, cl.hasOption(forceOpt.getOpt())), false);
return 0;
Modified: accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/InsertCommand.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/InsertCommand.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/InsertCommand.java (original)
+++ accumulo/trunk/core/src/main/java/org/apache/accumulo/core/util/shell/commands/InsertCommand.java Tue Sep 18 14:25:37 2012
@@ -22,6 +22,7 @@ import java.util.ArrayList;
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.MutationsRejectedException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.ConstraintViolationSummary;
@@ -62,7 +63,8 @@ public class InsertCommand extends Comma
else
m.put(colf, colq, val);
- final BatchWriter bw = shellState.getConnector().createBatchWriter(shellState.getTableName(), m.estimatedMemoryUsed() + 0L, 0L, 1);
+ final BatchWriter bw = shellState.getConnector().createBatchWriter(shellState.getTableName(),
+ new BatchWriterConfig().setMaxMemory(m.estimatedMemoryUsed()).setMaxWriteThreads(1));
bw.addMutation(m);
try {
bw.close();
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/ClientSideIteratorTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/ClientSideIteratorTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/ClientSideIteratorTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/ClientSideIteratorTest.java Tue Sep 18 14:25:37 2012
@@ -66,7 +66,7 @@ public class ClientSideIteratorTest {
Instance instance = new MockInstance("local");
Connector conn = instance.getConnector("root", "".getBytes());
conn.tableOperations().create("intersect");
- BatchWriter bw = conn.createBatchWriter("intersect", 1000, 10l, 1);
+ BatchWriter bw = conn.createBatchWriter("intersect", new BatchWriterConfig());
Mutation m = new Mutation("part1");
m.put("bar", "doc1", "value");
m.put("bar", "doc2", "value");
@@ -99,7 +99,7 @@ public class ClientSideIteratorTest {
conn.tableOperations().removeProperty("table", "table.iterator.scan.vers");
conn.tableOperations().removeProperty("table", "table.iterator.majc.vers");
conn.tableOperations().removeProperty("table", "table.iterator.minc.vers");
- final BatchWriter bw = conn.createBatchWriter("table", 1000, 10l, 1);
+ final BatchWriter bw = conn.createBatchWriter("table", new BatchWriterConfig());
Mutation m = new Mutation("row1");
m.put("colf", "colq", 1l, "value");
m.put("colf", "colq", 2l, "value");
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/admin/FindMaxTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/admin/FindMaxTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/admin/FindMaxTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/admin/FindMaxTest.java Tue Sep 18 14:25:37 2012
@@ -23,6 +23,7 @@ import junit.framework.TestCase;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.mock.MockInstance;
@@ -51,7 +52,7 @@ public class FindMaxTest extends TestCas
Connector conn = mi.getConnector("root", "foo");
conn.tableOperations().create("foo");
- BatchWriter bw = conn.createBatchWriter("foo", 1000000, 60000l, 2);
+ BatchWriter bw = conn.createBatchWriter("foo", new BatchWriterConfig());
bw.addMutation(nm(new byte[] {0}));
bw.addMutation(nm(new byte[] {0, 0}));
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java Tue Sep 18 14:25:37 2012
@@ -26,6 +26,7 @@ import java.util.HashSet;
import java.util.List;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.mapreduce.InputFormatBase.RangeInputSplit;
@@ -248,7 +249,7 @@ public class AccumuloInputFormatTest {
MockInstance mockInstance = new MockInstance("testmapinstance");
Connector c = mockInstance.getConnector("root", new byte[] {});
c.tableOperations().create("testtable");
- BatchWriter bw = c.createBatchWriter("testtable", 10000L, 1000L, 4);
+ BatchWriter bw = c.createBatchWriter("testtable", new BatchWriterConfig());
for (int i = 0; i < 100; i++) {
Mutation m = new Mutation(new Text(String.format("%09x", i + 1)));
m.put(new Text(), new Text(), new Value(String.format("%09x", i).getBytes()));
@@ -287,7 +288,7 @@ public class AccumuloInputFormatTest {
MockInstance mockInstance = new MockInstance("testmapinstance");
Connector c = mockInstance.getConnector("root", new byte[] {});
c.tableOperations().create("testtable2");
- BatchWriter bw = c.createBatchWriter("testtable2", 10000L, 1000L, 4);
+ BatchWriter bw = c.createBatchWriter("testtable2", new BatchWriterConfig());
for (int i = 0; i < 100; i++) {
Mutation m = new Mutation(new Text(String.format("%09x", i + 1)));
m.put(new Text(), new Text(), new Value(String.format("%09x", i).getBytes()));
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java Tue Sep 18 14:25:37 2012
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map.Entry;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.mock.MockInstance;
@@ -78,7 +79,7 @@ public class AccumuloOutputFormatTest {
Connector c = mockInstance.getConnector("root", new byte[] {});
c.tableOperations().create("testtable1");
c.tableOperations().create("testtable2");
- BatchWriter bw = c.createBatchWriter("testtable1", 10000L, 1000L, 4);
+ BatchWriter bw = c.createBatchWriter("testtable1", new BatchWriterConfig());
for (int i = 0; i < 100; i++) {
Mutation m = new Mutation(new Text(String.format("%09x", i + 1)));
m.put(new Text(), new Text(), new Value(String.format("%09x", i).getBytes()));
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloRowInputFormatTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloRowInputFormatTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloRowInputFormatTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloRowInputFormatTest.java Tue Sep 18 14:25:37 2012
@@ -29,6 +29,7 @@ import java.util.Map.Entry;
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.Connector;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.TableExistsException;
@@ -104,7 +105,7 @@ public class AccumuloRowInputFormatTest
conn.tableOperations().create("test");
BatchWriter writer = null; // NOPMD
try {
- writer = conn.createBatchWriter("test", 100000l, 100l, 5);
+ writer = conn.createBatchWriter("test", new BatchWriterConfig());
insertList(writer, row1);
insertList(writer, row2);
insertList(writer, row3);
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/client/mock/MockConnectorTest.java Tue Sep 18 14:25:37 2012
@@ -29,6 +29,7 @@ import org.apache.accumulo.core.Constant
import org.apache.accumulo.core.client.BatchDeleter;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
@@ -54,7 +55,7 @@ public class MockConnectorTest {
public void testSunnyDay() throws Exception {
Connector c = new MockConnector("root");
c.tableOperations().create("test");
- BatchWriter bw = c.createBatchWriter("test", 10000L, 1000L, 4);
+ BatchWriter bw = c.createBatchWriter("test", new BatchWriterConfig());
for (int i = 0; i < 100; i++) {
int r = random.nextInt();
Mutation m = new Mutation(asText(r));
@@ -98,7 +99,7 @@ public class MockConnectorTest {
c.tableOperations().attachIterator(table, is);
String keys[][] = { {"foo", "day", "20080101"}, {"foo", "day", "20080101"}, {"foo", "day", "20080103"}, {"bar", "day", "20080101"},
{"bar", "day", "20080101"},};
- BatchWriter bw = c.createBatchWriter("perDayCounts", 1000L, 1000L, 1);
+ BatchWriter bw = c.createBatchWriter("perDayCounts", new BatchWriterConfig());
for (String elt[] : keys) {
Mutation m = new Mutation(new Text(elt[0]));
m.put(new Text(elt[1]), new Text(elt[2]), new Value("1".getBytes()));
@@ -121,7 +122,7 @@ public class MockConnectorTest {
public void testDelete() throws Exception {
Connector c = new MockConnector("root");
c.tableOperations().create("test");
- BatchWriter bw = c.createBatchWriter("test", 10000L, 1000L, 4);
+ BatchWriter bw = c.createBatchWriter("test", new BatchWriterConfig());
Mutation m1 = new Mutation("r1");
@@ -165,14 +166,14 @@ public class MockConnectorTest {
c.tableOperations().delete("test");
c.tableOperations().create("test");
- BatchDeleter deleter = c.createBatchDeleter("test", Constants.NO_AUTHS, 2, 10000L, 1000L, 4);
+ BatchDeleter deleter = c.createBatchDeleter("test", Constants.NO_AUTHS, 2, new BatchWriterConfig());
// first make sure it deletes fine when its empty
deleter.setRanges(Collections.singletonList(new Range(("r1"))));
deleter.delete();
this.checkRemaining(c, "test", 0);
// test deleting just one row
- BatchWriter writer = c.createBatchWriter("test", 10, 10, 1);
+ BatchWriter writer = c.createBatchWriter("test", new BatchWriterConfig());
Mutation m = new Mutation("r1");
m.put("fam", "qual", "value");
writer.addMutation(m);
@@ -186,7 +187,7 @@ public class MockConnectorTest {
this.checkRemaining(c, "test", 0);
// test multi row deletes
- writer = c.createBatchWriter("test", 10, 10, 1);
+ writer = c.createBatchWriter("test", new BatchWriterConfig());
m = new Mutation("r1");
m.put("fam", "qual", "value");
writer.addMutation(m);
@@ -231,7 +232,7 @@ public class MockConnectorTest {
// test writing to a table that the is being scanned
Connector c = new MockConnector("root");
c.tableOperations().create("test");
- BatchWriter bw = c.createBatchWriter("test", 10000L, 1000L, 4);
+ BatchWriter bw = c.createBatchWriter("test", new BatchWriterConfig());
for (int i = 0; i < 10; i++) {
Mutation m1 = new Mutation("r" + i);
@@ -283,7 +284,7 @@ public class MockConnectorTest {
Connector c = new MockConnector("root");
c.tableOperations().create("a");
c.tableOperations().create("b");
- MultiTableBatchWriter bw = c.createMultiTableBatchWriter(10000L, 1000L, 4);
+ MultiTableBatchWriter bw = c.createMultiTableBatchWriter(new BatchWriterConfig());
Mutation m1 = new Mutation("r1");
m1.put("cf1", "cq1", 1, "v1");
BatchWriter b = bw.getBatchWriter("a");
@@ -314,7 +315,7 @@ public class MockConnectorTest {
public void testUpdate() throws Exception {
Connector c = new MockConnector("root");
c.tableOperations().create("test");
- BatchWriter bw = c.createBatchWriter("test", 1000, 1000l, 1);
+ BatchWriter bw = c.createBatchWriter("test", new BatchWriterConfig());
for (int i = 0; i < 10; i++) {
Mutation m = new Mutation("r1");
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/IntersectingIteratorTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/IntersectingIteratorTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/IntersectingIteratorTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/IntersectingIteratorTest.java Tue Sep 18 14:25:37 2012
@@ -32,6 +32,7 @@ import junit.framework.TestCase;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.mock.MockInstance;
@@ -274,7 +275,7 @@ public class IntersectingIteratorTest ex
MockInstance inst = new MockInstance("mockabye");
Connector connector = inst.getConnector("user", "pass");
connector.tableOperations().create("index");
- BatchWriter bw = connector.createBatchWriter("index", 1000, 1000, 1);
+ BatchWriter bw = connector.createBatchWriter("index", new BatchWriterConfig());
Mutation m = new Mutation("000012");
m.put("rvy", "5000000000000000", empty);
m.put("15qh", "5000000000000000", empty);
Modified: accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/RowFilterTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/RowFilterTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/RowFilterTest.java (original)
+++ accumulo/trunk/core/src/test/java/org/apache/accumulo/core/iterators/user/RowFilterTest.java Tue Sep 18 14:25:37 2012
@@ -25,6 +25,7 @@ import junit.framework.TestCase;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.Scanner;
@@ -84,7 +85,7 @@ public class RowFilterTest extends TestC
Connector conn = instance.getConnector("", "".getBytes());
conn.tableOperations().create("table1");
- BatchWriter bw = conn.createBatchWriter("table1", 1000000, 60000, 1);
+ BatchWriter bw = conn.createBatchWriter("table1", new BatchWriterConfig());
Mutation m = new Mutation("0");
m.put("cf1", "cq1", "1");
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RandomBatchWriter.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RandomBatchWriter.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RandomBatchWriter.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RandomBatchWriter.java Tue Sep 18 14:25:37 2012
@@ -18,10 +18,12 @@ package org.apache.accumulo.examples.sim
import java.util.HashSet;
import java.util.Random;
+import java.util.concurrent.TimeUnit;
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.Connector;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.TableNotFoundException;
@@ -142,7 +144,8 @@ public class RandomBatchWriter {
ZooKeeperInstance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector connector = instance.getConnector(user, pass);
- BatchWriter bw = connector.createBatchWriter(table, maxMemory, maxLatency, numThreads);
+ BatchWriter bw = connector.createBatchWriter(table, new BatchWriterConfig().setMaxMemory(maxMemory).setMaxLatency(maxLatency, TimeUnit.MILLISECONDS)
+ .setMaxWriteThreads(numThreads));
// reuse the ColumnVisibility object to improve performance
ColumnVisibility cv = new ColumnVisibility(visiblity);
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/ReadWriteExample.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/ReadWriteExample.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/ReadWriteExample.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/ReadWriteExample.java Tue Sep 18 14:25:37 2012
@@ -24,6 +24,7 @@ import org.apache.accumulo.core.Constant
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.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.MutationsRejectedException;
@@ -165,7 +166,7 @@ public class ReadWriteExample {
}
private void createEntries(boolean delete) throws AccumuloException, TableNotFoundException, MutationsRejectedException {
- BatchWriter writer = conn.createBatchWriter(getOpt(tableNameOpt, DEFAULT_TABLE_NAME), 10000, Long.MAX_VALUE, 1);
+ BatchWriter writer = conn.createBatchWriter(getOpt(tableNameOpt, DEFAULT_TABLE_NAME), new BatchWriterConfig());
ColumnVisibility cv = new ColumnVisibility(DEFAULT_AUTHS.replace(',', '|'));
Text cf = new Text("datatypes");
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RowOperations.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RowOperations.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RowOperations.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/RowOperations.java Tue Sep 18 14:25:37 2012
@@ -23,6 +23,7 @@ import org.apache.accumulo.core.Constant
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.Connector;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
@@ -93,7 +94,7 @@ public class RowOperations {
mut3.put(new Text("column"), col4, System.currentTimeMillis(), new Value("This is the value for this key".getBytes()));
// Now we'll make a Batch Writer
- bw = connector.createBatchWriter(table, 100000l, 30l, 1);
+ bw = connector.createBatchWriter(table, new BatchWriterConfig());
// And add the mutations
bw.addMutation(mut1);
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/SequentialBatchWriter.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/SequentialBatchWriter.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/SequentialBatchWriter.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/client/SequentialBatchWriter.java Tue Sep 18 14:25:37 2012
@@ -16,9 +16,12 @@
*/
package org.apache.accumulo.examples.simple.client;
+import java.util.concurrent.TimeUnit;
+
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.Connector;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.TableNotFoundException;
@@ -65,7 +68,8 @@ public class SequentialBatchWriter {
ZooKeeperInstance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector connector = instance.getConnector(user, pass);
- BatchWriter bw = connector.createBatchWriter(table, maxMemory, maxLatency, numThreads);
+ BatchWriter bw = connector.createBatchWriter(table, new BatchWriterConfig().setMaxMemory(maxMemory).setMaxLatency(maxLatency, TimeUnit.MILLISECONDS)
+ .setMaxWriteThreads(numThreads));
long end = start + num;
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/FileCount.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/FileCount.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/FileCount.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/FileCount.java Tue Sep 18 14:25:37 2012
@@ -20,6 +20,7 @@ import java.util.Iterator;
import java.util.Map.Entry;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.Scanner;
@@ -256,7 +257,7 @@ public class FileCount {
inserts = 0;
Scanner scanner = conn.createScanner(table, auths);
- BatchWriter bw = conn.createBatchWriter(table, 10000000, 60000l, 3);
+ BatchWriter bw = conn.createBatchWriter(table, new BatchWriterConfig());
long t1 = System.currentTimeMillis();
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/Ingest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/Ingest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/Ingest.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/dirlist/Ingest.java Tue Sep 18 14:25:37 2012
@@ -20,6 +20,7 @@ import java.io.File;
import java.io.IOException;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.ZooKeeperInstance;
@@ -142,9 +143,9 @@ public class Ingest {
conn.tableOperations().attachIterator(dataTable, new IteratorSetting(1, ChunkCombiner.class));
}
- BatchWriter dirBW = conn.createBatchWriter(nameTable, 50000000, 300000l, 4);
- BatchWriter indexBW = conn.createBatchWriter(indexTable, 50000000, 300000l, 4);
- BatchWriter dataBW = conn.createBatchWriter(dataTable, 50000000, 300000l, 4);
+ BatchWriter dirBW = conn.createBatchWriter(nameTable, new BatchWriterConfig());
+ BatchWriter indexBW = conn.createBatchWriter(indexTable, new BatchWriterConfig());
+ BatchWriter dataBW = conn.createBatchWriter(dataTable, new BatchWriterConfig());
FileDataIngest fdi = new FileDataIngest(chunkSize, colvis);
for (int i = 9; i < args.length; i++) {
recurse(new File(args[i]), colvis, dirBW, indexBW, fdi, dataBW);
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataIngest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataIngest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataIngest.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/filedata/FileDataIngest.java Tue Sep 18 14:25:37 2012
@@ -23,6 +23,7 @@ import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.MutationsRejectedException;
@@ -185,7 +186,7 @@ public class FileDataIngest {
conn.tableOperations().create(dataTable);
conn.tableOperations().attachIterator(dataTable, new IteratorSetting(1, ChunkCombiner.class));
}
- BatchWriter bw = conn.createBatchWriter(dataTable, 50000000, 300000l, 4);
+ BatchWriter bw = conn.createBatchWriter(dataTable, new BatchWriterConfig());
FileDataIngest fdi = new FileDataIngest(chunkSize, colvis);
for (int i = 7; i < args.length; i++) {
fdi.insertFileData(args[i], bw);
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/helloworld/InsertWithBatchWriter.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/helloworld/InsertWithBatchWriter.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/helloworld/InsertWithBatchWriter.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/helloworld/InsertWithBatchWriter.java Tue Sep 18 14:25:37 2012
@@ -19,6 +19,7 @@ package org.apache.accumulo.examples.sim
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.Connector;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
import org.apache.accumulo.core.client.MutationsRejectedException;
@@ -49,7 +50,7 @@ public class InsertWithBatchWriter {
ZooKeeperInstance instance = new ZooKeeperInstance(instanceName, zooKeepers);
Connector connector = instance.getConnector(user, pass);
- MultiTableBatchWriter mtbw = connector.createMultiTableBatchWriter(200000l, 300, 4);
+ MultiTableBatchWriter mtbw = connector.createMultiTableBatchWriter(new BatchWriterConfig());
BatchWriter bw = null;
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/isolation/InterferenceTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/isolation/InterferenceTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/isolation/InterferenceTest.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/isolation/InterferenceTest.java Tue Sep 18 14:25:37 2012
@@ -21,6 +21,7 @@ import java.util.Map.Entry;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.IsolatedScanner;
import org.apache.accumulo.core.client.MutationsRejectedException;
@@ -158,7 +159,7 @@ public class InterferenceTest {
if (!conn.tableOperations().exists(table))
conn.tableOperations().create(table);
- Thread writer = new Thread(new Writer(conn.createBatchWriter(table, 10000000, 60000l, 3)));
+ Thread writer = new Thread(new Writer(conn.createBatchWriter(table, new BatchWriterConfig())));
writer.start();
Reader r;
if (Boolean.parseBoolean(args[6]))
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Index.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Index.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Index.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Index.java Tue Sep 18 14:25:37 2012
@@ -21,6 +21,7 @@ import java.io.FileReader;
import java.util.HashSet;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.ZooKeeperInstance;
import org.apache.accumulo.core.data.Mutation;
@@ -91,7 +92,7 @@ public class Index {
private static BatchWriter setupBatchWriter(String instance, String zooKeepers, String table, String user, String pass) throws Exception {
ZooKeeperInstance zinstance = new ZooKeeperInstance(instance, zooKeepers);
Connector conn = zinstance.getConnector(user, pass.getBytes());
- return conn.createBatchWriter(table, 50000000, 300000l, 4);
+ return conn.createBatchWriter(table, new BatchWriterConfig());
}
public static void main(String[] args) throws Exception {
Modified: accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Reverse.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Reverse.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Reverse.java (original)
+++ accumulo/trunk/examples/simple/src/main/java/org/apache/accumulo/examples/simple/shard/Reverse.java Tue Sep 18 14:25:37 2012
@@ -20,6 +20,7 @@ import java.util.Map.Entry;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.ZooKeeperInstance;
@@ -54,7 +55,7 @@ public class Reverse {
Connector conn = zki.getConnector(user, pass.getBytes());
Scanner scanner = conn.createScanner(inTable, Constants.NO_AUTHS);
- BatchWriter bw = conn.createBatchWriter(outTable, 50000000, 600000l, 4);
+ BatchWriter bw = conn.createBatchWriter(outTable, new BatchWriterConfig());
for (Entry<Key,Value> entry : scanner) {
Key key = entry.getKey();
Modified: accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/dirlist/CountTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/dirlist/CountTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/dirlist/CountTest.java (original)
+++ accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/dirlist/CountTest.java Tue Sep 18 14:25:37 2012
@@ -22,6 +22,7 @@ import java.util.Map.Entry;
import junit.framework.TestCase;
import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.mock.MockInstance;
@@ -30,9 +31,6 @@ import org.apache.accumulo.core.data.Val
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.accumulo.core.util.Pair;
-import org.apache.accumulo.examples.simple.dirlist.FileCount;
-import org.apache.accumulo.examples.simple.dirlist.Ingest;
-import org.apache.accumulo.examples.simple.dirlist.QueryUtil;
import org.apache.hadoop.io.Text;
public class CountTest extends TestCase {
@@ -40,7 +38,7 @@ public class CountTest extends TestCase
try {
Connector conn = new MockInstance("counttest").getConnector("root", "".getBytes());
conn.tableOperations().create("dirlisttable");
- BatchWriter bw = conn.createBatchWriter("dirlisttable", 1000000l, 100l, 1);
+ BatchWriter bw = conn.createBatchWriter("dirlisttable", new BatchWriterConfig());
ColumnVisibility cv = new ColumnVisibility();
// / has 1 dir
// /local has 2 dirs 1 file
Modified: accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputFormatTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputFormatTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputFormatTest.java (original)
+++ accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputFormatTest.java Tue Sep 18 14:25:37 2012
@@ -27,6 +27,7 @@ import junit.framework.TestCase;
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.Connector;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
@@ -75,7 +76,7 @@ public class ChunkInputFormatTest extend
MockInstance instance = new MockInstance("instance1");
Connector conn = instance.getConnector("root", "".getBytes());
conn.tableOperations().create("test");
- BatchWriter bw = conn.createBatchWriter("test", 100000l, 100l, 5);
+ BatchWriter bw = conn.createBatchWriter("test", new BatchWriterConfig());
for (Entry<Key,Value> e : data) {
Key k = e.getKey();
@@ -126,7 +127,7 @@ public class ChunkInputFormatTest extend
MockInstance instance = new MockInstance("instance2");
Connector conn = instance.getConnector("root", "".getBytes());
conn.tableOperations().create("test");
- BatchWriter bw = conn.createBatchWriter("test", 100000l, 100l, 5);
+ BatchWriter bw = conn.createBatchWriter("test", new BatchWriterConfig());
for (Entry<Key,Value> e : data) {
Key k = e.getKey();
@@ -166,7 +167,7 @@ public class ChunkInputFormatTest extend
MockInstance instance = new MockInstance("instance3");
Connector conn = instance.getConnector("root", "".getBytes());
conn.tableOperations().create("test");
- BatchWriter bw = conn.createBatchWriter("test", 100000l, 100l, 5);
+ BatchWriter bw = conn.createBatchWriter("test", new BatchWriterConfig());
for (Entry<Key,Value> e : baddata) {
Key k = e.getKey();
Mutation m = new Mutation(k.getRow());
Modified: accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputStreamTest.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputStreamTest.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputStreamTest.java (original)
+++ accumulo/trunk/examples/simple/src/test/java/org/apache/accumulo/examples/simple/filedata/ChunkInputStreamTest.java Tue Sep 18 14:25:37 2012
@@ -26,6 +26,7 @@ import junit.framework.TestCase;
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.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableExistsException;
@@ -222,7 +223,7 @@ public class ChunkInputStreamTest extend
public void testWithAccumulo() throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException, IOException {
Connector conn = new MockInstance().getConnector("root", "".getBytes());
conn.tableOperations().create("test");
- BatchWriter bw = conn.createBatchWriter("test", 100000l, 100l, 5);
+ BatchWriter bw = conn.createBatchWriter("test", new BatchWriterConfig());
for (Entry<Key,Value> e : data) {
Key k = e.getKey();
Modified: accumulo/trunk/server/src/main/java/org/apache/accumulo/server/gc/SimpleGarbageCollector.java
URL: http://svn.apache.org/viewvc/accumulo/trunk/server/src/main/java/org/apache/accumulo/server/gc/SimpleGarbageCollector.java?rev=1387182&r1=1387181&r2=1387182&view=diff
==============================================================================
--- accumulo/trunk/server/src/main/java/org/apache/accumulo/server/gc/SimpleGarbageCollector.java (original)
+++ accumulo/trunk/server/src/main/java/org/apache/accumulo/server/gc/SimpleGarbageCollector.java Tue Sep 18 14:25:37 2012
@@ -41,6 +41,7 @@ import org.apache.accumulo.core.Constant
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.Connector;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.IsolatedScanner;
@@ -550,7 +551,7 @@ public class SimpleGarbageCollector impl
Connector c;
try {
c = instance.getConnector(SecurityConstants.getSystemCredentials());
- writer = c.createBatchWriter(Constants.METADATA_TABLE_NAME, 10000000, 60000l, 3);
+ writer = c.createBatchWriter(Constants.METADATA_TABLE_NAME, new BatchWriterConfig());
} catch (Exception e) {
log.error("Unable to create writer to remove file from the !METADATA table", e);
}