@Test public void consumedButNotSentDataShouldBeRecorded() throws Exception { Client replClient = createMock(Client.class); AccumuloReplicaSystem ars = createMock(AccumuloReplicaSystem.class); WalEdits edits = new WalEdits(Collections.emptyList()); WalReplication walReplication = new WalReplication(edits, 0, 5, 0); ReplicationTarget target = new ReplicationTarget("peer", "2", TableId.of("1")); DataInputStream input = null; Path p = new Path("/accumulo/wals/tserver+port/" + UUID.randomUUID()); Status status = null; long sizeLimit = Long.MAX_VALUE; String remoteTableId = target.getRemoteIdentifier(); TCredentials tcreds = null; Set<Integer> tids = new HashSet<>(); WalClientExecReturn walClientExec = ars.new WalClientExecReturn(target, input, p, status, sizeLimit, remoteTableId, tcreds, tids); expect(ars.getWalEdits(target, input, p, status, sizeLimit, tids)).andReturn(walReplication); replay(replClient, ars); ReplicationStats stats = walClientExec.execute(replClient); verify(replClient, ars); assertEquals(new ReplicationStats(0L, 0L, 5L), stats); }
protected WalReplication getWalEdits(ReplicationTarget target, DataInputStream wal, Path p, Status status, long sizeLimit, Set<Integer> desiredTids) throws IOException { WalEdits edits = new WalEdits(); edits.edits = new ArrayList<>(); long size = 0L; long entriesConsumed = 0L; long numUpdates = 0L; LogFileKey key = new LogFileKey(); LogFileValue value = new LogFileValue(); while (size < sizeLimit) { try { key.readFields(wal); value.readFields(wal); } catch (EOFException e) { log.debug("Caught EOFException reading {}", p); if (status.getInfiniteEnd() && status.getClosed()) { log.debug("{} is closed and has unknown length, assuming entire file has been consumed", p); entriesConsumed = Long.MAX_VALUE; } break; } entriesConsumed++; switch (key.event) { case DEFINE_TABLET: if (target.getSourceTableId().equals(key.tablet.tableId())) { desiredTids.add(key.tabletId); } break; case MUTATION: case MANY_MUTATIONS: if (desiredTids.contains(key.tabletId)) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos); key.write(out); numUpdates += writeValueAvoidingReplicationCycles(out, value, target); out.flush(); byte[] data = baos.toByteArray(); size += data.length; edits.addToEdits(ByteBuffer.wrap(data)); } break; default: log.trace("Ignorning WAL entry which doesn't contain mutations," + " should not have received such entries"); break; } } return new WalReplication(edits, size, entriesConsumed, numUpdates); }
AccumuloReplicaSystem implements ReplicaSystem { protected WalReplication getWalEdits(ReplicationTarget target, DataInputStream wal, Path p, Status status, long sizeLimit, Set<Integer> desiredTids) throws IOException { WalEdits edits = new WalEdits(); edits.edits = new ArrayList<>(); long size = 0L; long entriesConsumed = 0L; long numUpdates = 0L; LogFileKey key = new LogFileKey(); LogFileValue value = new LogFileValue(); while (size < sizeLimit) { try { key.readFields(wal); value.readFields(wal); } catch (EOFException e) { log.debug("Caught EOFException reading {}", p); if (status.getInfiniteEnd() && status.getClosed()) { log.debug("{} is closed and has unknown length, assuming entire file has been consumed", p); entriesConsumed = Long.MAX_VALUE; } break; } entriesConsumed++; switch (key.event) { case DEFINE_TABLET: if (target.getSourceTableId().equals(key.tablet.tableId())) { desiredTids.add(key.tabletId); } break; case MUTATION: case MANY_MUTATIONS: if (desiredTids.contains(key.tabletId)) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos); key.write(out); numUpdates += writeValueAvoidingReplicationCycles(out, value, target); out.flush(); byte[] data = baos.toByteArray(); size += data.length; edits.addToEdits(ByteBuffer.wrap(data)); } break; default: log.trace("Ignorning WAL entry which doesn't contain mutations," + " should not have received such entries"); break; } } return new WalReplication(edits, size, entriesConsumed, numUpdates); } }
AccumuloReplicaSystem implements ReplicaSystem { protected WalReplication getWalEdits(ReplicationTarget target, DataInputStream wal, Path p, Status status, long sizeLimit, Set<Integer> desiredTids) throws IOException { WalEdits edits = new WalEdits(); edits.edits = new ArrayList<>(); long size = 0L; long entriesConsumed = 0L; long numUpdates = 0L; LogFileKey key = new LogFileKey(); LogFileValue value = new LogFileValue(); while (size < sizeLimit) { try { key.readFields(wal); value.readFields(wal); } catch (EOFException e) { log.debug("Caught EOFException reading {}", p); if (status.getInfiniteEnd() && status.getClosed()) { log.debug("{} is closed and has unknown length, assuming entire file has been consumed", p); entriesConsumed = Long.MAX_VALUE; } break; } entriesConsumed++; switch (key.event) { case DEFINE_TABLET: if (target.getSourceTableId().equals(key.tablet.tableId())) { desiredTids.add(key.tabletId); } break; case MUTATION: case MANY_MUTATIONS: if (desiredTids.contains(key.tabletId)) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos); key.write(out); numUpdates += writeValueAvoidingReplicationCycles(out, value, target); out.flush(); byte[] data = baos.toByteArray(); size += data.length; edits.addToEdits(ByteBuffer.wrap(data)); } break; default: log.trace("Ignorning WAL entry which doesn't contain mutations," + " should not have received such entries"); break; } } return new WalReplication(edits, size, entriesConsumed, numUpdates); }  }
AccumuloReplicaSystem implements ReplicaSystem { protected WalReplication getWalEdits(ReplicationTarget target, DataInputStream wal, Path p, Status status, long sizeLimit, Set<Integer> desiredTids) throws IOException { WalEdits edits = new WalEdits(); edits.edits = new ArrayList<>(); long size = 0L; long entriesConsumed = 0L; long numUpdates = 0L; LogFileKey key = new LogFileKey(); LogFileValue value = new LogFileValue(); while (size < sizeLimit) { try { key.readFields(wal); value.readFields(wal); } catch (EOFException e) { log.debug("Caught EOFException reading {}", p); if (status.getInfiniteEnd() && status.getClosed()) { log.debug("{} is closed and has unknown length, assuming entire file has been consumed", p); entriesConsumed = Long.MAX_VALUE; } break; } entriesConsumed++; switch (key.event) { case DEFINE_TABLET: if (target.getSourceTableId().equals(key.tablet.tableId())) { desiredTids.add(key.tabletId); } break; case MUTATION: case MANY_MUTATIONS: if (desiredTids.contains(key.tabletId)) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos); key.write(out); numUpdates += writeValueAvoidingReplicationCycles(out, value, target); out.flush(); byte[] data = baos.toByteArray(); size += data.length; edits.addToEdits(ByteBuffer.wrap(data)); } break; default: log.trace("Ignorning WAL entry which doesn't contain mutations," + " should not have received such entries"); break; } } return new WalReplication(edits, size, entriesConsumed, numUpdates); }  static String buildConfiguration(String instanceName, String zookeepers); @Override void configure(ServerContext context, String configuration); @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "path provided by admin") @Override Status replicate(final Path p, final Status status, final ReplicationTarget target,
      final ReplicaSystemHelper helper); DataInputStream getWalStream(Path p, FSDataInputStream input); }
AccumuloReplicaSystem implements ReplicaSystem { protected WalReplication getWalEdits(ReplicationTarget target, DataInputStream wal, Path p, Status status, long sizeLimit, Set<Integer> desiredTids) throws IOException { WalEdits edits = new WalEdits(); edits.edits = new ArrayList<>(); long size = 0L; long entriesConsumed = 0L; long numUpdates = 0L; LogFileKey key = new LogFileKey(); LogFileValue value = new LogFileValue(); while (size < sizeLimit) { try { key.readFields(wal); value.readFields(wal); } catch (EOFException e) { log.debug("Caught EOFException reading {}", p); if (status.getInfiniteEnd() && status.getClosed()) { log.debug("{} is closed and has unknown length, assuming entire file has been consumed", p); entriesConsumed = Long.MAX_VALUE; } break; } entriesConsumed++; switch (key.event) { case DEFINE_TABLET: if (target.getSourceTableId().equals(key.tablet.tableId())) { desiredTids.add(key.tabletId); } break; case MUTATION: case MANY_MUTATIONS: if (desiredTids.contains(key.tabletId)) { ByteArrayOutputStream baos = new ByteArrayOutputStream(); DataOutputStream out = new DataOutputStream(baos); key.write(out); numUpdates += writeValueAvoidingReplicationCycles(out, value, target); out.flush(); byte[] data = baos.toByteArray(); size += data.length; edits.addToEdits(ByteBuffer.wrap(data)); } break; default: log.trace("Ignorning WAL entry which doesn't contain mutations," + " should not have received such entries"); break; } } return new WalReplication(edits, size, entriesConsumed, numUpdates); }  static String buildConfiguration(String instanceName, String zookeepers); @Override void configure(ServerContext context, String configuration); @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "path provided by admin") @Override Status replicate(final Path p, final Status status, final ReplicationTarget target,
      final ReplicaSystemHelper helper); DataInputStream getWalStream(Path p, FSDataInputStream input);  }