@Test public void systemTimestampsAreSetOnUpdates() throws Exception { final BatchWriterReplicationReplayer replayer = new BatchWriterReplicationReplayer(); final String tableName = "foo"; final long systemTimestamp = 1000; final BatchWriterConfig bwCfg = new BatchWriterConfig(); bwCfg.setMaxMemory(1L); LogFileKey key = new LogFileKey(); key.event = LogEvents.MANY_MUTATIONS; key.seq = 1; key.tabletId = 1; WalEdits edits = new WalEdits(); Mutation m = new Mutation("row"); m.put("cf", "cq1", "value"); m.put("cf", "cq2", "value"); m.put("cf", "cq3", "value"); m.put("cf", "cq4", "value"); m.put("cf", "cq5", "value"); TMutation tMutation = m.toThrift(); ServerMutation sMutation = new ServerMutation(tMutation); sMutation.setSystemTimestamp(systemTimestamp); ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos); key.write(out); out.writeInt(1); sMutation.write(out); out.close(); edits.addToEdits(ByteBuffer.wrap(baos.toByteArray())); Mutation expectedMutation = new Mutation("row"); expectedMutation.put("cf", "cq1", sMutation.getSystemTimestamp(), "value"); expectedMutation.put("cf", "cq2", sMutation.getSystemTimestamp(), "value"); expectedMutation.put("cf", "cq3", sMutation.getSystemTimestamp(), "value"); expectedMutation.put("cf", "cq4", sMutation.getSystemTimestamp(), "value"); expectedMutation.put("cf", "cq5", sMutation.getSystemTimestamp(), "value"); expect(conf.getAsBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY)) .andReturn(bwCfg.getMaxMemory()); expect(context.createBatchWriter(tableName, bwCfg)).andReturn(bw); bw.addMutations(Lists.newArrayList(expectedMutation)); expectLastCall().once(); bw.close(); expectLastCall().once(); replay(context, conf, bw); replayer.replicateLog(context, tableName, edits); }
@Override public long replicateLog(ClientContext context, String tableName, WalEdits data) throws RemoteReplicationException { final LogFileKey key = new LogFileKey(); final LogFileValue value = new LogFileValue(); final long memoryInBytes = context.getConfiguration().getAsBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY); BatchWriter bw = null; long mutationsApplied = 0L; try { for (ByteBuffer edit : data.getEdits()) { DataInputStream dis = new DataInputStream(ByteBufferUtil.toByteArrayInputStream(edit)); try { key.readFields(dis); value.readFields(dis); } catch (IOException e) { log.error("Could not deserialize edit from stream", e); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_DESERIALIZE, "Could not deserialize edit from stream"); } if (bw == null) { BatchWriterConfig bwConfig = new BatchWriterConfig(); bwConfig.setMaxMemory(memoryInBytes); try { bw = context.createBatchWriter(tableName, bwConfig); } catch (TableNotFoundException e) { throw new RemoteReplicationException(RemoteReplicationErrorCode.TABLE_DOES_NOT_EXIST, "Table " + tableName + " does not exist"); } } log.info("Applying {} mutations to table {} as part of batch", value.mutations.size(), tableName); List<Mutation> mutationsCopy = new ArrayList<>(value.mutations.size()); long mutationsCopied = 0L; for (Mutation orig : value.mutations) { if (orig instanceof ServerMutation) { mutationsCopied++; ServerMutation origServer = (ServerMutation) orig; Mutation copy = new Mutation(orig.getRow()); for (ColumnUpdate update : orig.getUpdates()) { long timestamp; if (update.hasTimestamp()) { timestamp = update.getTimestamp(); } else { timestamp = origServer.getSystemTimestamp(); } if (update.isDeleted()) { copy.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp); } else { copy.put(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp, update.getValue()); } } Set<String> replicationSources = orig.getReplicationSources(); if (replicationSources != null && !replicationSources.isEmpty()) { for (String replicationSource : replicationSources) { copy.addReplicationSource(replicationSource); } } mutationsCopy.add(copy); } else { mutationsCopy.add(orig); } } log.debug("Copied {} mutations to ensure server-assigned timestamps are propagated", mutationsCopied); try { bw.addMutations(mutationsCopy); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } log.debug("{} mutations added to the BatchScanner", mutationsCopy.size()); mutationsApplied += mutationsCopy.size(); } } finally { if (bw != null) { try { bw.close(); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } } } log.info("Applied {} mutations in total to {}", mutationsApplied, tableName); return mutationsApplied; }
BatchWriterReplicationReplayer implements AccumuloReplicationReplayer { @Override public long replicateLog(ClientContext context, String tableName, WalEdits data) throws RemoteReplicationException { final LogFileKey key = new LogFileKey(); final LogFileValue value = new LogFileValue(); final long memoryInBytes = context.getConfiguration().getAsBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY); BatchWriter bw = null; long mutationsApplied = 0L; try { for (ByteBuffer edit : data.getEdits()) { DataInputStream dis = new DataInputStream(ByteBufferUtil.toByteArrayInputStream(edit)); try { key.readFields(dis); value.readFields(dis); } catch (IOException e) { log.error("Could not deserialize edit from stream", e); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_DESERIALIZE, "Could not deserialize edit from stream"); } if (bw == null) { BatchWriterConfig bwConfig = new BatchWriterConfig(); bwConfig.setMaxMemory(memoryInBytes); try { bw = context.createBatchWriter(tableName, bwConfig); } catch (TableNotFoundException e) { throw new RemoteReplicationException(RemoteReplicationErrorCode.TABLE_DOES_NOT_EXIST, "Table " + tableName + " does not exist"); } } log.info("Applying {} mutations to table {} as part of batch", value.mutations.size(), tableName); List<Mutation> mutationsCopy = new ArrayList<>(value.mutations.size()); long mutationsCopied = 0L; for (Mutation orig : value.mutations) { if (orig instanceof ServerMutation) { mutationsCopied++; ServerMutation origServer = (ServerMutation) orig; Mutation copy = new Mutation(orig.getRow()); for (ColumnUpdate update : orig.getUpdates()) { long timestamp; if (update.hasTimestamp()) { timestamp = update.getTimestamp(); } else { timestamp = origServer.getSystemTimestamp(); } if (update.isDeleted()) { copy.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp); } else { copy.put(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp, update.getValue()); } } Set<String> replicationSources = orig.getReplicationSources(); if (replicationSources != null && !replicationSources.isEmpty()) { for (String replicationSource : replicationSources) { copy.addReplicationSource(replicationSource); } } mutationsCopy.add(copy); } else { mutationsCopy.add(orig); } } log.debug("Copied {} mutations to ensure server-assigned timestamps are propagated", mutationsCopied); try { bw.addMutations(mutationsCopy); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } log.debug("{} mutations added to the BatchScanner", mutationsCopy.size()); mutationsApplied += mutationsCopy.size(); } } finally { if (bw != null) { try { bw.close(); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } } } log.info("Applied {} mutations in total to {}", mutationsApplied, tableName); return mutationsApplied; } }
BatchWriterReplicationReplayer implements AccumuloReplicationReplayer { @Override public long replicateLog(ClientContext context, String tableName, WalEdits data) throws RemoteReplicationException { final LogFileKey key = new LogFileKey(); final LogFileValue value = new LogFileValue(); final long memoryInBytes = context.getConfiguration().getAsBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY); BatchWriter bw = null; long mutationsApplied = 0L; try { for (ByteBuffer edit : data.getEdits()) { DataInputStream dis = new DataInputStream(ByteBufferUtil.toByteArrayInputStream(edit)); try { key.readFields(dis); value.readFields(dis); } catch (IOException e) { log.error("Could not deserialize edit from stream", e); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_DESERIALIZE, "Could not deserialize edit from stream"); } if (bw == null) { BatchWriterConfig bwConfig = new BatchWriterConfig(); bwConfig.setMaxMemory(memoryInBytes); try { bw = context.createBatchWriter(tableName, bwConfig); } catch (TableNotFoundException e) { throw new RemoteReplicationException(RemoteReplicationErrorCode.TABLE_DOES_NOT_EXIST, "Table " + tableName + " does not exist"); } } log.info("Applying {} mutations to table {} as part of batch", value.mutations.size(), tableName); List<Mutation> mutationsCopy = new ArrayList<>(value.mutations.size()); long mutationsCopied = 0L; for (Mutation orig : value.mutations) { if (orig instanceof ServerMutation) { mutationsCopied++; ServerMutation origServer = (ServerMutation) orig; Mutation copy = new Mutation(orig.getRow()); for (ColumnUpdate update : orig.getUpdates()) { long timestamp; if (update.hasTimestamp()) { timestamp = update.getTimestamp(); } else { timestamp = origServer.getSystemTimestamp(); } if (update.isDeleted()) { copy.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp); } else { copy.put(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp, update.getValue()); } } Set<String> replicationSources = orig.getReplicationSources(); if (replicationSources != null && !replicationSources.isEmpty()) { for (String replicationSource : replicationSources) { copy.addReplicationSource(replicationSource); } } mutationsCopy.add(copy); } else { mutationsCopy.add(orig); } } log.debug("Copied {} mutations to ensure server-assigned timestamps are propagated", mutationsCopied); try { bw.addMutations(mutationsCopy); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } log.debug("{} mutations added to the BatchScanner", mutationsCopy.size()); mutationsApplied += mutationsCopy.size(); } } finally { if (bw != null) { try { bw.close(); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } } } log.info("Applied {} mutations in total to {}", mutationsApplied, tableName); return mutationsApplied; }  }
BatchWriterReplicationReplayer implements AccumuloReplicationReplayer { @Override public long replicateLog(ClientContext context, String tableName, WalEdits data) throws RemoteReplicationException { final LogFileKey key = new LogFileKey(); final LogFileValue value = new LogFileValue(); final long memoryInBytes = context.getConfiguration().getAsBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY); BatchWriter bw = null; long mutationsApplied = 0L; try { for (ByteBuffer edit : data.getEdits()) { DataInputStream dis = new DataInputStream(ByteBufferUtil.toByteArrayInputStream(edit)); try { key.readFields(dis); value.readFields(dis); } catch (IOException e) { log.error("Could not deserialize edit from stream", e); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_DESERIALIZE, "Could not deserialize edit from stream"); } if (bw == null) { BatchWriterConfig bwConfig = new BatchWriterConfig(); bwConfig.setMaxMemory(memoryInBytes); try { bw = context.createBatchWriter(tableName, bwConfig); } catch (TableNotFoundException e) { throw new RemoteReplicationException(RemoteReplicationErrorCode.TABLE_DOES_NOT_EXIST, "Table " + tableName + " does not exist"); } } log.info("Applying {} mutations to table {} as part of batch", value.mutations.size(), tableName); List<Mutation> mutationsCopy = new ArrayList<>(value.mutations.size()); long mutationsCopied = 0L; for (Mutation orig : value.mutations) { if (orig instanceof ServerMutation) { mutationsCopied++; ServerMutation origServer = (ServerMutation) orig; Mutation copy = new Mutation(orig.getRow()); for (ColumnUpdate update : orig.getUpdates()) { long timestamp; if (update.hasTimestamp()) { timestamp = update.getTimestamp(); } else { timestamp = origServer.getSystemTimestamp(); } if (update.isDeleted()) { copy.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp); } else { copy.put(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp, update.getValue()); } } Set<String> replicationSources = orig.getReplicationSources(); if (replicationSources != null && !replicationSources.isEmpty()) { for (String replicationSource : replicationSources) { copy.addReplicationSource(replicationSource); } } mutationsCopy.add(copy); } else { mutationsCopy.add(orig); } } log.debug("Copied {} mutations to ensure server-assigned timestamps are propagated", mutationsCopied); try { bw.addMutations(mutationsCopy); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } log.debug("{} mutations added to the BatchScanner", mutationsCopy.size()); mutationsApplied += mutationsCopy.size(); } } finally { if (bw != null) { try { bw.close(); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } } } log.info("Applied {} mutations in total to {}", mutationsApplied, tableName); return mutationsApplied; }  @Override long replicateLog(ClientContext context, String tableName, WalEdits data); }
BatchWriterReplicationReplayer implements AccumuloReplicationReplayer { @Override public long replicateLog(ClientContext context, String tableName, WalEdits data) throws RemoteReplicationException { final LogFileKey key = new LogFileKey(); final LogFileValue value = new LogFileValue(); final long memoryInBytes = context.getConfiguration().getAsBytes(Property.TSERV_REPLICATION_BW_REPLAYER_MEMORY); BatchWriter bw = null; long mutationsApplied = 0L; try { for (ByteBuffer edit : data.getEdits()) { DataInputStream dis = new DataInputStream(ByteBufferUtil.toByteArrayInputStream(edit)); try { key.readFields(dis); value.readFields(dis); } catch (IOException e) { log.error("Could not deserialize edit from stream", e); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_DESERIALIZE, "Could not deserialize edit from stream"); } if (bw == null) { BatchWriterConfig bwConfig = new BatchWriterConfig(); bwConfig.setMaxMemory(memoryInBytes); try { bw = context.createBatchWriter(tableName, bwConfig); } catch (TableNotFoundException e) { throw new RemoteReplicationException(RemoteReplicationErrorCode.TABLE_DOES_NOT_EXIST, "Table " + tableName + " does not exist"); } } log.info("Applying {} mutations to table {} as part of batch", value.mutations.size(), tableName); List<Mutation> mutationsCopy = new ArrayList<>(value.mutations.size()); long mutationsCopied = 0L; for (Mutation orig : value.mutations) { if (orig instanceof ServerMutation) { mutationsCopied++; ServerMutation origServer = (ServerMutation) orig; Mutation copy = new Mutation(orig.getRow()); for (ColumnUpdate update : orig.getUpdates()) { long timestamp; if (update.hasTimestamp()) { timestamp = update.getTimestamp(); } else { timestamp = origServer.getSystemTimestamp(); } if (update.isDeleted()) { copy.putDelete(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp); } else { copy.put(update.getColumnFamily(), update.getColumnQualifier(), new ColumnVisibility(update.getColumnVisibility()), timestamp, update.getValue()); } } Set<String> replicationSources = orig.getReplicationSources(); if (replicationSources != null && !replicationSources.isEmpty()) { for (String replicationSource : replicationSources) { copy.addReplicationSource(replicationSource); } } mutationsCopy.add(copy); } else { mutationsCopy.add(orig); } } log.debug("Copied {} mutations to ensure server-assigned timestamps are propagated", mutationsCopied); try { bw.addMutations(mutationsCopy); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } log.debug("{} mutations added to the BatchScanner", mutationsCopy.size()); mutationsApplied += mutationsCopy.size(); } } finally { if (bw != null) { try { bw.close(); } catch (MutationsRejectedException e) { log.error("Could not apply mutations to {}", tableName); throw new RemoteReplicationException(RemoteReplicationErrorCode.COULD_NOT_APPLY, "Could not apply mutations to " + tableName); } } } log.info("Applied {} mutations in total to {}", mutationsApplied, tableName); return mutationsApplied; }  @Override long replicateLog(ClientContext context, String tableName, WalEdits data);  }