Index: hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java =================================================================== --- hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java (revision 1511890) +++ hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java (working copy) @@ -20,14 +20,13 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.TreeMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; +import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; @@ -45,14 +44,12 @@ import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.HConnection; import org.apache.hadoop.hbase.client.HConnectionManager; -import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.HTableInterface; import org.apache.hadoop.hbase.client.Mutation; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Row; import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry; import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos; -import org.apache.hadoop.hbase.util.Threads; /** * This class is responsible for replicating the edits coming @@ -75,7 +72,6 @@ // Name of the HDFS directory that contains the temporary rep logs public static final String REPLICATION_LOG_DIR = ".replogs"; private final Configuration conf; - private final ExecutorService sharedThreadPool; private final HConnection sharedHtableCon; private final MetricsSink metrics; private final AtomicLong totalReplicatedEdits = new AtomicLong(); @@ -93,11 +89,6 @@ decorateConf(); this.metrics = new MetricsSink(); this.sharedHtableCon = HConnectionManager.createConnection(this.conf); - this.sharedThreadPool = new ThreadPoolExecutor(1, - conf.getInt("hbase.htable.threads.max", Integer.MAX_VALUE), - conf.getLong("hbase.htable.threads.keepalivetime", 60), TimeUnit.SECONDS, - new SynchronousQueue(), Threads.newDaemonThreadFactory("hbase-repl")); - ((ThreadPoolExecutor) this.sharedThreadPool).allowCoreThreadTimeOut(true); } /** @@ -125,10 +116,9 @@ // to the same table. try { long totalReplicated = 0; - // Map of table => list of Rows, we only want to flushCommits once per - // invocation of this method per table. - Map> rows = - new TreeMap>(); + // Map of table => list of Rows, grouped by cluster id, we only want to flushCommits once per + // invocation of this method per table and cluster id. + Map>> rowMap = new TreeMap>>(); for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); @@ -148,7 +138,7 @@ new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()): new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); m.setClusterId(uuid); - addToMultiMap(rows, table, m); + addToHashMultiMap(rowMap, table, uuid, m); } if (CellUtil.isDelete(cell)) { ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell)); @@ -159,8 +149,8 @@ } totalReplicated++; } - for (Entry> entry : rows.entrySet()) { - batch(entry.getKey(), entry.getValue()); + for (Entry>> entry : rowMap.entrySet()) { + batch(entry.getKey(), entry.getValue().values()); } int size = entries.size(); this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime()); @@ -190,15 +180,21 @@ * Simple helper to a map from key to (a list of) values * TODO: Make a general utility method * @param map - * @param key + * @param key1 + * @param key2 * @param value * @return */ - private List addToMultiMap(Map> map, K key, V value) { - List values = map.get(key); + private List addToHashMultiMap(Map>> map, K1 key1, K2 key2, V value) { + Map> innerMap = map.get(key1); + if (innerMap == null) { + innerMap = new HashMap>(); + map.put(key1, innerMap); + } + List values = innerMap.get(key2); if (values == null) { values = new ArrayList(); - map.put(key, values); + innerMap.put(key2, values); } values.add(value); return values; @@ -209,15 +205,6 @@ */ public void stopReplicationSinkServices() { try { - this.sharedThreadPool.shutdown(); - if (!this.sharedThreadPool.awaitTermination(60000, TimeUnit.MILLISECONDS)) { - this.sharedThreadPool.shutdownNow(); - } - } catch (InterruptedException e) { - LOG.warn("Interrupted while closing the table pool", e); // ignoring it as we are closing. - Thread.currentThread().interrupt(); - } - try { this.sharedHtableCon.close(); } catch (IOException e) { LOG.warn("IOException while closing the connection", e); // ignoring as we are closing. @@ -231,14 +218,16 @@ * @param rows list of actions * @throws IOException */ - private void batch(TableName tableName, List rows) throws IOException { - if (rows.isEmpty()) { + private void batch(TableName tableName, Collection> allRows) throws IOException { + if (allRows.isEmpty()) { return; } HTableInterface table = null; try { - table = new HTable(tableName, this.sharedHtableCon, this.sharedThreadPool); - table.batch(rows); + table = this.sharedHtableCon.getTable(tableName); + for (List rows : allRows) { + table.batch(rows); + } } catch (InterruptedException ix) { throw new IOException(ix); } finally {