@Test public void testMixedDeletes() throws Exception { HLog.Entry[] entries = new HLog.Entry[3]; for(int i = 0; i < 3; i++) { entries[i] = createEntry(TABLE_NAME1, i, KeyValue.Type.Put); } SINK.replicateEntries(entries); entries = new HLog.Entry[3]; entries[0] = createEntry(TABLE_NAME1, 0, KeyValue.Type.DeleteColumn); entries[1] = createEntry(TABLE_NAME1, 1, KeyValue.Type.DeleteFamily); entries[2] = createEntry(TABLE_NAME1, 2, KeyValue.Type.DeleteColumn); SINK.replicateEntries(entries); Scan scan = new Scan(); ResultScanner scanRes = table1.getScanner(scan); assertEquals(0, scanRes.next(3).length); }
public void replicateEntries(HLog.Entry[] entries) throws IOException { if (entries.length == 0) { return; } try { long totalReplicated = 0; Map<byte[], List<Put>> puts = new TreeMap<byte[], List<Put>>(Bytes.BYTES_COMPARATOR); for (HLog.Entry entry : entries) { WALEdit edit = entry.getEdit(); List<KeyValue> kvs = edit.getKeyValues(); if (kvs.get(0).isDelete()) { Delete delete = new Delete(kvs.get(0).getRow(), kvs.get(0).getTimestamp(), null); for (KeyValue kv : kvs) { if (kv.isDeleteFamily()) { delete.deleteFamily(kv.getFamily()); } else if (!kv.isEmptyColumn()) { delete.deleteColumn(kv.getFamily(), kv.getQualifier()); } } delete(entry.getKey().getTablename(), delete); } else { byte[] table = entry.getKey().getTablename(); List<Put> tableList = puts.get(table); if (tableList == null) { tableList = new ArrayList<Put>(); puts.put(table, tableList); } byte[] lastKey = kvs.get(0).getRow(); Put put = new Put(kvs.get(0).getRow(), kvs.get(0).getTimestamp()); for (KeyValue kv : kvs) { if (!Bytes.equals(lastKey, kv.getRow())) { tableList.add(put); put = new Put(kv.getRow(), kv.getTimestamp()); } put.add(kv.getFamily(), kv.getQualifier(), kv.getValue()); lastKey = kv.getRow(); } tableList.add(put); } totalReplicated++; } for(byte [] table : puts.keySet()) { put(table, puts.get(table)); } this.metrics.setAgeOfLastAppliedOp( entries[entries.length-1].getKey().getWriteTime()); this.metrics.appliedBatchesRate.inc(1); LOG.info("Total replicated: " + totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } }
ReplicationSink { public void replicateEntries(HLog.Entry[] entries) throws IOException { if (entries.length == 0) { return; } try { long totalReplicated = 0; Map<byte[], List<Put>> puts = new TreeMap<byte[], List<Put>>(Bytes.BYTES_COMPARATOR); for (HLog.Entry entry : entries) { WALEdit edit = entry.getEdit(); List<KeyValue> kvs = edit.getKeyValues(); if (kvs.get(0).isDelete()) { Delete delete = new Delete(kvs.get(0).getRow(), kvs.get(0).getTimestamp(), null); for (KeyValue kv : kvs) { if (kv.isDeleteFamily()) { delete.deleteFamily(kv.getFamily()); } else if (!kv.isEmptyColumn()) { delete.deleteColumn(kv.getFamily(), kv.getQualifier()); } } delete(entry.getKey().getTablename(), delete); } else { byte[] table = entry.getKey().getTablename(); List<Put> tableList = puts.get(table); if (tableList == null) { tableList = new ArrayList<Put>(); puts.put(table, tableList); } byte[] lastKey = kvs.get(0).getRow(); Put put = new Put(kvs.get(0).getRow(), kvs.get(0).getTimestamp()); for (KeyValue kv : kvs) { if (!Bytes.equals(lastKey, kv.getRow())) { tableList.add(put); put = new Put(kv.getRow(), kv.getTimestamp()); } put.add(kv.getFamily(), kv.getQualifier(), kv.getValue()); lastKey = kv.getRow(); } tableList.add(put); } totalReplicated++; } for(byte [] table : puts.keySet()) { put(table, puts.get(table)); } this.metrics.setAgeOfLastAppliedOp( entries[entries.length-1].getKey().getWriteTime()); this.metrics.appliedBatchesRate.inc(1); LOG.info("Total replicated: " + totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } }
ReplicationSink { public void replicateEntries(HLog.Entry[] entries) throws IOException { if (entries.length == 0) { return; } try { long totalReplicated = 0; Map<byte[], List<Put>> puts = new TreeMap<byte[], List<Put>>(Bytes.BYTES_COMPARATOR); for (HLog.Entry entry : entries) { WALEdit edit = entry.getEdit(); List<KeyValue> kvs = edit.getKeyValues(); if (kvs.get(0).isDelete()) { Delete delete = new Delete(kvs.get(0).getRow(), kvs.get(0).getTimestamp(), null); for (KeyValue kv : kvs) { if (kv.isDeleteFamily()) { delete.deleteFamily(kv.getFamily()); } else if (!kv.isEmptyColumn()) { delete.deleteColumn(kv.getFamily(), kv.getQualifier()); } } delete(entry.getKey().getTablename(), delete); } else { byte[] table = entry.getKey().getTablename(); List<Put> tableList = puts.get(table); if (tableList == null) { tableList = new ArrayList<Put>(); puts.put(table, tableList); } byte[] lastKey = kvs.get(0).getRow(); Put put = new Put(kvs.get(0).getRow(), kvs.get(0).getTimestamp()); for (KeyValue kv : kvs) { if (!Bytes.equals(lastKey, kv.getRow())) { tableList.add(put); put = new Put(kv.getRow(), kv.getTimestamp()); } put.add(kv.getFamily(), kv.getQualifier(), kv.getValue()); lastKey = kv.getRow(); } tableList.add(put); } totalReplicated++; } for(byte [] table : puts.keySet()) { put(table, puts.get(table)); } this.metrics.setAgeOfLastAppliedOp( entries[entries.length-1].getKey().getWriteTime()); this.metrics.appliedBatchesRate.inc(1); LOG.info("Total replicated: " + totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } ReplicationSink(Configuration conf, Stoppable stopper); }
ReplicationSink { public void replicateEntries(HLog.Entry[] entries) throws IOException { if (entries.length == 0) { return; } try { long totalReplicated = 0; Map<byte[], List<Put>> puts = new TreeMap<byte[], List<Put>>(Bytes.BYTES_COMPARATOR); for (HLog.Entry entry : entries) { WALEdit edit = entry.getEdit(); List<KeyValue> kvs = edit.getKeyValues(); if (kvs.get(0).isDelete()) { Delete delete = new Delete(kvs.get(0).getRow(), kvs.get(0).getTimestamp(), null); for (KeyValue kv : kvs) { if (kv.isDeleteFamily()) { delete.deleteFamily(kv.getFamily()); } else if (!kv.isEmptyColumn()) { delete.deleteColumn(kv.getFamily(), kv.getQualifier()); } } delete(entry.getKey().getTablename(), delete); } else { byte[] table = entry.getKey().getTablename(); List<Put> tableList = puts.get(table); if (tableList == null) { tableList = new ArrayList<Put>(); puts.put(table, tableList); } byte[] lastKey = kvs.get(0).getRow(); Put put = new Put(kvs.get(0).getRow(), kvs.get(0).getTimestamp()); for (KeyValue kv : kvs) { if (!Bytes.equals(lastKey, kv.getRow())) { tableList.add(put); put = new Put(kv.getRow(), kv.getTimestamp()); } put.add(kv.getFamily(), kv.getQualifier(), kv.getValue()); lastKey = kv.getRow(); } tableList.add(put); } totalReplicated++; } for(byte [] table : puts.keySet()) { put(table, puts.get(table)); } this.metrics.setAgeOfLastAppliedOp( entries[entries.length-1].getKey().getWriteTime()); this.metrics.appliedBatchesRate.inc(1); LOG.info("Total replicated: " + totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } ReplicationSink(Configuration conf, Stoppable stopper); void replicateEntries(HLog.Entry[] entries); }
ReplicationSink { public void replicateEntries(HLog.Entry[] entries) throws IOException { if (entries.length == 0) { return; } try { long totalReplicated = 0; Map<byte[], List<Put>> puts = new TreeMap<byte[], List<Put>>(Bytes.BYTES_COMPARATOR); for (HLog.Entry entry : entries) { WALEdit edit = entry.getEdit(); List<KeyValue> kvs = edit.getKeyValues(); if (kvs.get(0).isDelete()) { Delete delete = new Delete(kvs.get(0).getRow(), kvs.get(0).getTimestamp(), null); for (KeyValue kv : kvs) { if (kv.isDeleteFamily()) { delete.deleteFamily(kv.getFamily()); } else if (!kv.isEmptyColumn()) { delete.deleteColumn(kv.getFamily(), kv.getQualifier()); } } delete(entry.getKey().getTablename(), delete); } else { byte[] table = entry.getKey().getTablename(); List<Put> tableList = puts.get(table); if (tableList == null) { tableList = new ArrayList<Put>(); puts.put(table, tableList); } byte[] lastKey = kvs.get(0).getRow(); Put put = new Put(kvs.get(0).getRow(), kvs.get(0).getTimestamp()); for (KeyValue kv : kvs) { if (!Bytes.equals(lastKey, kv.getRow())) { tableList.add(put); put = new Put(kv.getRow(), kv.getTimestamp()); } put.add(kv.getFamily(), kv.getQualifier(), kv.getValue()); lastKey = kv.getRow(); } tableList.add(put); } totalReplicated++; } for(byte [] table : puts.keySet()) { put(table, puts.get(table)); } this.metrics.setAgeOfLastAppliedOp( entries[entries.length-1].getKey().getWriteTime()); this.metrics.appliedBatchesRate.inc(1); LOG.info("Total replicated: " + totalReplicated); } catch (IOException ex) { LOG.error("Unable to accept edit because:", ex); throw ex; } } ReplicationSink(Configuration conf, Stoppable stopper); void replicateEntries(HLog.Entry[] entries); static final String REPLICATION_LOG_DIR; }