@Test public void testMixedPutTables() throws Exception { List<WALEntry> entries = new ArrayList<WALEntry>(BATCH_SIZE/2); List<Cell> cells = new ArrayList<Cell>(); for(int i = 0; i < BATCH_SIZE; i++) { entries.add(createEntry( i % 2 == 0 ? TABLE_NAME2 : TABLE_NAME1, i, KeyValue.Type.Put, cells)); } SINK.replicateEntries(entries, CellUtil.createCellScanner(cells.iterator())); Scan scan = new Scan(); ResultScanner scanRes = table2.getScanner(scan); for(Result res : scanRes) { assertTrue(Bytes.toInt(res.getRow()) % 2 == 0); } }
public void replicateEntries(List<WALEntry> entries, final CellScanner cells) throws IOException { if (entries.isEmpty()) return; if (cells == null) throw new NullPointerException("TODO: Add handling of null CellScanner"); try { long totalReplicated = 0; Map<TableName, Map<List<UUID>, List<Row>>> rowMap = new TreeMap<TableName, Map<List<UUID>, List<Row>>>(); for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); Cell previousCell = null; Mutation m = null; int count = entry.getAssociatedCellCount(); for (int i = 0; i < count; i++) { if (!cells.advance()) { throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i); } Cell cell = cells.current(); if (isNewRowOrType(previousCell, cell)) { m = CellUtil.isDelete(cell)? new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()): new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); List<UUID> clusterIds = new ArrayList<UUID>(); for(HBaseProtos.UUID clusterId : entry.getKey().getClusterIdsList()){ clusterIds.add(toUUID(clusterId)); } m.setClusterIds(clusterIds); addToHashMultiMap(rowMap, table, clusterIds, m); } if (CellUtil.isDelete(cell)) { ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell)); } else { ((Put)m).add(KeyValueUtil.ensureKeyValue(cell)); } previousCell = cell; } totalReplicated++; } for (Entry<TableName, Map<List<UUID>,List<Row>>> entry : rowMap.entrySet()) { batch(entry.getKey(), entry.getValue().values()); } int size = entries.size(); this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime()); this.metrics.applyBatch(size); this.totalReplicatedEdits.addAndGet(totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } }
ReplicationSink { public void replicateEntries(List<WALEntry> entries, final CellScanner cells) throws IOException { if (entries.isEmpty()) return; if (cells == null) throw new NullPointerException("TODO: Add handling of null CellScanner"); try { long totalReplicated = 0; Map<TableName, Map<List<UUID>, List<Row>>> rowMap = new TreeMap<TableName, Map<List<UUID>, List<Row>>>(); for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); Cell previousCell = null; Mutation m = null; int count = entry.getAssociatedCellCount(); for (int i = 0; i < count; i++) { if (!cells.advance()) { throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i); } Cell cell = cells.current(); if (isNewRowOrType(previousCell, cell)) { m = CellUtil.isDelete(cell)? new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()): new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); List<UUID> clusterIds = new ArrayList<UUID>(); for(HBaseProtos.UUID clusterId : entry.getKey().getClusterIdsList()){ clusterIds.add(toUUID(clusterId)); } m.setClusterIds(clusterIds); addToHashMultiMap(rowMap, table, clusterIds, m); } if (CellUtil.isDelete(cell)) { ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell)); } else { ((Put)m).add(KeyValueUtil.ensureKeyValue(cell)); } previousCell = cell; } totalReplicated++; } for (Entry<TableName, Map<List<UUID>,List<Row>>> entry : rowMap.entrySet()) { batch(entry.getKey(), entry.getValue().values()); } int size = entries.size(); this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime()); this.metrics.applyBatch(size); this.totalReplicatedEdits.addAndGet(totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } }
ReplicationSink { public void replicateEntries(List<WALEntry> entries, final CellScanner cells) throws IOException { if (entries.isEmpty()) return; if (cells == null) throw new NullPointerException("TODO: Add handling of null CellScanner"); try { long totalReplicated = 0; Map<TableName, Map<List<UUID>, List<Row>>> rowMap = new TreeMap<TableName, Map<List<UUID>, List<Row>>>(); for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); Cell previousCell = null; Mutation m = null; int count = entry.getAssociatedCellCount(); for (int i = 0; i < count; i++) { if (!cells.advance()) { throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i); } Cell cell = cells.current(); if (isNewRowOrType(previousCell, cell)) { m = CellUtil.isDelete(cell)? new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()): new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); List<UUID> clusterIds = new ArrayList<UUID>(); for(HBaseProtos.UUID clusterId : entry.getKey().getClusterIdsList()){ clusterIds.add(toUUID(clusterId)); } m.setClusterIds(clusterIds); addToHashMultiMap(rowMap, table, clusterIds, m); } if (CellUtil.isDelete(cell)) { ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell)); } else { ((Put)m).add(KeyValueUtil.ensureKeyValue(cell)); } previousCell = cell; } totalReplicated++; } for (Entry<TableName, Map<List<UUID>,List<Row>>> entry : rowMap.entrySet()) { batch(entry.getKey(), entry.getValue().values()); } int size = entries.size(); this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime()); this.metrics.applyBatch(size); this.totalReplicatedEdits.addAndGet(totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } ReplicationSink(Configuration conf, Stoppable stopper); }
ReplicationSink { public void replicateEntries(List<WALEntry> entries, final CellScanner cells) throws IOException { if (entries.isEmpty()) return; if (cells == null) throw new NullPointerException("TODO: Add handling of null CellScanner"); try { long totalReplicated = 0; Map<TableName, Map<List<UUID>, List<Row>>> rowMap = new TreeMap<TableName, Map<List<UUID>, List<Row>>>(); for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); Cell previousCell = null; Mutation m = null; int count = entry.getAssociatedCellCount(); for (int i = 0; i < count; i++) { if (!cells.advance()) { throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i); } Cell cell = cells.current(); if (isNewRowOrType(previousCell, cell)) { m = CellUtil.isDelete(cell)? new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()): new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); List<UUID> clusterIds = new ArrayList<UUID>(); for(HBaseProtos.UUID clusterId : entry.getKey().getClusterIdsList()){ clusterIds.add(toUUID(clusterId)); } m.setClusterIds(clusterIds); addToHashMultiMap(rowMap, table, clusterIds, m); } if (CellUtil.isDelete(cell)) { ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell)); } else { ((Put)m).add(KeyValueUtil.ensureKeyValue(cell)); } previousCell = cell; } totalReplicated++; } for (Entry<TableName, Map<List<UUID>,List<Row>>> entry : rowMap.entrySet()) { batch(entry.getKey(), entry.getValue().values()); } int size = entries.size(); this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime()); this.metrics.applyBatch(size); this.totalReplicatedEdits.addAndGet(totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } ReplicationSink(Configuration conf, Stoppable stopper); void replicateEntries(List<WALEntry> entries, final CellScanner cells); void stopReplicationSinkServices(); String getStats(); }
ReplicationSink { public void replicateEntries(List<WALEntry> entries, final CellScanner cells) throws IOException { if (entries.isEmpty()) return; if (cells == null) throw new NullPointerException("TODO: Add handling of null CellScanner"); try { long totalReplicated = 0; Map<TableName, Map<List<UUID>, List<Row>>> rowMap = new TreeMap<TableName, Map<List<UUID>, List<Row>>>(); for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); Cell previousCell = null; Mutation m = null; int count = entry.getAssociatedCellCount(); for (int i = 0; i < count; i++) { if (!cells.advance()) { throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i); } Cell cell = cells.current(); if (isNewRowOrType(previousCell, cell)) { m = CellUtil.isDelete(cell)? new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()): new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()); List<UUID> clusterIds = new ArrayList<UUID>(); for(HBaseProtos.UUID clusterId : entry.getKey().getClusterIdsList()){ clusterIds.add(toUUID(clusterId)); } m.setClusterIds(clusterIds); addToHashMultiMap(rowMap, table, clusterIds, m); } if (CellUtil.isDelete(cell)) { ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell)); } else { ((Put)m).add(KeyValueUtil.ensureKeyValue(cell)); } previousCell = cell; } totalReplicated++; } for (Entry<TableName, Map<List<UUID>,List<Row>>> entry : rowMap.entrySet()) { batch(entry.getKey(), entry.getValue().values()); } int size = entries.size(); this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime()); this.metrics.applyBatch(size); this.totalReplicatedEdits.addAndGet(totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } ReplicationSink(Configuration conf, Stoppable stopper); void replicateEntries(List<WALEntry> entries, final CellScanner cells); void stopReplicationSinkServices(); String getStats();  }