Java 类org.apache.hadoop.hbase.wal.WAL.Entry 实例源码

项目：ditb 文件：ScopeWALEntryFilter.java

@Override
public Entry filter(Entry entry) {
  NavigableMap<byte[], Integer> scopes = entry.getKey().getScopes();
  if (scopes == null || scopes.isEmpty()) {
    return null;
  }
  ArrayList<Cell> cells = entry.getEdit().getCells();
  int size = cells.size();
  for (int i = size - 1; i >= 0; i--) {
    Cell cell = cells.get(i);
    // The scope will be null or empty if
    // there's nothing to replicate in that WALEdit
    if (!scopes.containsKey(cell.getFamily())
        || scopes.get(cell.getFamily()) == HConstants.REPLICATION_SCOPE_LOCAL) {
      cells.remove(i);
    }
  }
  if (cells.size() < size / 2) {
    cells.trimToSize();
  }
  return entry;
}

项目：ditb 文件：HBaseInterClusterReplicationEndpoint.java

@Override
public Integer call() throws IOException {
  SinkPeer sinkPeer = null;
  try {
    sinkPeer = replicationSinkMgr.getReplicationSink();
    BlockingInterface rrs = sinkPeer.getRegionServer();
    ReplicationProtbufUtil.replicateWALEntry(rrs,
        entries.toArray(new Entry[entries.size()]));
    replicationSinkMgr.reportSinkSuccess(sinkPeer);
    return ordinal;

  } catch (IOException ioe) {
    if (sinkPeer != null) {
      replicationSinkMgr.reportBadSink(sinkPeer);
    }
    throw ioe;
  }
}

项目：ditb 文件：RegionReplicaReplicationEndpoint.java

@Override
public void append(RegionEntryBuffer buffer) throws IOException {
  List<Entry> entries = buffer.getEntryBuffer();

  if (entries.isEmpty() || entries.get(0).getEdit().getCells().isEmpty()) {
    return;
  }

  // meta edits (e.g. flush) are always replicated.
  // data edits (e.g. put) are replicated if the table requires them.
  if (!requiresReplication(buffer.getTableName(), entries)) {
    return;
  }

  sinkWriter.append(buffer.getTableName(), buffer.getEncodedRegionName(),
    entries.get(0).getEdit().getCells().get(0).getRow(), entries);
}

项目：ditb 文件：WALSplitter.java

/**
 * Append a log entry into the corresponding region buffer.
 * Blocks if the total heap usage has crossed the specified threshold.
 *
 * @throws InterruptedException
 * @throws IOException
 */
public void appendEntry(Entry entry) throws InterruptedException, IOException {
  WALKey key = entry.getKey();

  RegionEntryBuffer buffer;
  long incrHeap;
  synchronized (this) {
    buffer = buffers.get(key.getEncodedRegionName());
    if (buffer == null) {
      buffer = new RegionEntryBuffer(key.getTablename(), key.getEncodedRegionName());
      buffers.put(key.getEncodedRegionName(), buffer);
    }
    incrHeap= buffer.appendEntry(entry);
  }

  // If we crossed the chunk threshold, wait for more space to be available
  synchronized (controller.dataAvailable) {
    totalBuffered += incrHeap;
    while (totalBuffered > maxHeapUsage && controller.thrown.get() == null) {
      LOG.debug("Used " + totalBuffered + " bytes of buffered edits, waiting for IO threads...");
      controller.dataAvailable.wait(2000);
    }
    controller.dataAvailable.notifyAll();
  }
  controller.checkForErrors();
}

项目：ditb 文件：WALSplitter.java

/**
 * @return RegionEntryBuffer a buffer of edits to be written or replayed.
 */
synchronized RegionEntryBuffer getChunkToWrite() {
  long biggestSize = 0;
  byte[] biggestBufferKey = null;

  for (Map.Entry<byte[], RegionEntryBuffer> entry : buffers.entrySet()) {
    long size = entry.getValue().heapSize();
    if (size > biggestSize && (!currentlyWriting.contains(entry.getKey()))) {
      biggestSize = size;
      biggestBufferKey = entry.getKey();
    }
  }
  if (biggestBufferKey == null) {
    return null;
  }

  RegionEntryBuffer buffer = buffers.remove(biggestBufferKey);
  currentlyWriting.add(biggestBufferKey);
  return buffer;
}

项目：ditb 文件：WALSplitter.java

/**
 * Get a writer and path for a log starting at the given entry. This function is threadsafe so
 * long as multiple threads are always acting on different regions.
 * @return null if this region shouldn't output any logs
 */
private WriterAndPath getWriterAndPath(Entry entry) throws IOException {
  byte region[] = entry.getKey().getEncodedRegionName();
  WriterAndPath ret = (WriterAndPath) writers.get(region);
  if (ret != null) {
    return ret;
  }
  // If we already decided that this region doesn't get any output
  // we don't need to check again.
  if (blacklistedRegions.contains(region)) {
    return null;
  }
  ret = createWAP(region, entry, rootDir);
  if (ret == null) {
    blacklistedRegions.add(region);
    return null;
  }
  writers.put(region, ret);
  return ret;
}

项目：ditb 文件：WALSplitter.java

/**
 * @return a path with a write for that path. caller should close.
 */
private WriterAndPath createWAP(byte[] region, Entry entry, Path rootdir) throws IOException {
  Path regionedits = getRegionSplitEditsPath(fs, entry, rootdir, true);
  if (regionedits == null) {
    return null;
  }
  if (fs.exists(regionedits)) {
    LOG.warn("Found old edits file. It could be the "
        + "result of a previous failed split attempt. Deleting " + regionedits + ", length="
        + fs.getFileStatus(regionedits).getLen());
    if (!fs.delete(regionedits, false)) {
      LOG.warn("Failed delete of old " + regionedits);
    }
  }
  Writer w = createWriter(regionedits);
  LOG.debug("Creating writer path=" + regionedits);
  return new WriterAndPath(regionedits, w);
}

项目：ditb 文件：WALSplitter.java

private void processWorkItems(String key, List<Pair<HRegionLocation, Entry>> actions)
    throws IOException {
  RegionServerWriter rsw = null;

  long startTime = System.nanoTime();
  try {
    rsw = getRegionServerWriter(key);
    rsw.sink.replayEntries(actions);

    // Pass along summary statistics
    rsw.incrementEdits(actions.size());
    rsw.incrementNanoTime(System.nanoTime() - startTime);
  } catch (IOException e) {
    e = RemoteExceptionHandler.checkIOException(e);
    LOG.fatal(" Got while writing log entry to log", e);
    throw e;
  }
}

项目：ditb 文件：WALEditsReplaySink.java

private void replayEdits(final HRegionLocation regionLoc, final HRegionInfo regionInfo,
    final List<Entry> entries) throws IOException {
  try {
    RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(conf, null);
    ReplayServerCallable<ReplicateWALEntryResponse> callable =
        new ReplayServerCallable<ReplicateWALEntryResponse>(this.conn, this.tableName, regionLoc,
            regionInfo, entries);
    factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, this.replayTimeout);
  } catch (IOException ie) {
    if (skipErrors) {
      LOG.warn(HConstants.HREGION_EDITS_REPLAY_SKIP_ERRORS
          + "=true so continuing replayEdits with error:" + ie.getMessage());
    } else {
      throw ie;
    }
  }
}

项目：ditb 文件：WALEditsReplaySink.java

private void replayToServer(HRegionInfo regionInfo, List<Entry> entries)
    throws IOException, ServiceException {
  if (entries.isEmpty()) return;

  Entry[] entriesArray = new Entry[entries.size()];
  entriesArray = entries.toArray(entriesArray);
  AdminService.BlockingInterface remoteSvr = conn.getAdmin(getLocation().getServerName());

  Pair<AdminProtos.ReplicateWALEntryRequest, CellScanner> p =
      ReplicationProtbufUtil.buildReplicateWALEntryRequest(entriesArray);
  PayloadCarryingRpcController controller = rpcControllerFactory.newController(p.getSecond());
  try {
    remoteSvr.replay(controller, p.getFirst());
  } catch (ServiceException se) {
    throw ProtobufUtil.getRemoteException(se);
  }
}

项目：ditb 文件：WALEditsReplaySink.java

@Override
public void prepare(boolean reload) throws IOException {
  if (!reload) return;
  // relocate regions in case we have a new dead server or network hiccup
  // if not due to connection issue, the following code should run fast because it uses
  // cached location
  boolean skip = false;
  for (Entry entry : this.entries) {
    WALEdit edit = entry.getEdit();
    List<Cell> cells = edit.getCells();
    for (Cell cell : cells) {
      // filtering WAL meta entries
      setLocation(conn.locateRegion(tableName, cell.getRow()));
      skip = true;
      break;
    }
    // use first log entry to relocate region because all entries are for one region
    if (skip) break;
  }
}

项目：ditb 文件：TestReplicationEndpoint.java

@Override
public WALEntryFilter getWALEntryfilter() {
  return new ChainWALEntryFilter(super.getWALEntryfilter(), new WALEntryFilter() {
    @Override
    public Entry filter(Entry entry) {
      ArrayList<Cell> cells = entry.getEdit().getCells();
      int size = cells.size();
      for (int i = size-1; i >= 0; i--) {
        Cell cell = cells.get(i);
        if (!Bytes.equals(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
          row, 0, row.length)) {
          cells.remove(i);
        }
      }
      return entry;
    }
  });
}

项目：ditb 文件：TestReplicationWALEntryFilters.java

@Test
public void testSystemTableWALEntryFilter() {
  SystemTableWALEntryFilter filter = new SystemTableWALEntryFilter();

  // meta
  WALKey key1 = new WALKey( HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes(),
    TableName.META_TABLE_NAME);
  Entry metaEntry = new Entry(key1, null);

  assertNull(filter.filter(metaEntry));

  // ns table
  WALKey key2 = new WALKey(new byte[] {}, TableName.NAMESPACE_TABLE_NAME);
  Entry nsEntry = new Entry(key2, null);
  assertNull(filter.filter(nsEntry));

  // user table

  WALKey key3 = new WALKey(new byte[] {}, TableName.valueOf("foo"));
  Entry userEntry = new Entry(key3, null);

  assertEquals(userEntry, filter.filter(userEntry));
}

项目：ditb 文件：TestReplicationWALEntryFilters.java

private void assertEquals(Entry e1, Entry e2) {
  Assert.assertEquals(e1 == null, e2 == null);
  if (e1 == null) {
    return;
  }

  // do not compare WALKeys

  // compare kvs
  Assert.assertEquals(e1.getEdit() == null, e2.getEdit() == null);
  if (e1.getEdit() == null) {
    return;
  }
  List<Cell> cells1 = e1.getEdit().getCells();
  List<Cell> cells2 = e2.getEdit().getCells();
  Assert.assertEquals(cells1.size(), cells2.size());
  for (int i = 0; i < cells1.size(); i++) {
    KeyValue.COMPARATOR.compare(cells1.get(i), cells2.get(i));
  }
}

项目：ditb 文件：TestRegionReplicaReplicationEndpointNoMaster.java

private void replicateUsingCallable(ClusterConnection connection, Queue<Entry> entries)
    throws IOException, RuntimeException {
  Entry entry;
  while ((entry = entries.poll()) != null) {
    byte[] row = entry.getEdit().getCells().get(0).getRow();
    RegionLocations locations = connection.locateRegion(tableName, row, true, true);
    RegionReplicaReplayCallable callable = new RegionReplicaReplayCallable(connection,
      RpcControllerFactory.instantiate(connection.getConfiguration()),
      table.getName(), locations.getRegionLocation(1),
      locations.getRegionLocation(1).getRegionInfo(), row, Lists.newArrayList(entry),
      new AtomicLong());

    RpcRetryingCallerFactory factory = RpcRetryingCallerFactory.instantiate(
      connection.getConfiguration());
    factory.<ReplicateWALEntryResponse> newCaller().callWithRetries(callable, 10000);
  }
}

项目：pbase 文件：ScopeWALEntryFilter.java

@Override
public Entry filter(Entry entry) {
  NavigableMap<byte[], Integer> scopes = entry.getKey().getScopes();
  if (scopes == null || scopes.isEmpty()) {
    return null;
  }
  ArrayList<Cell> cells = entry.getEdit().getCells();
  int size = cells.size();
  for (int i = size - 1; i >= 0; i--) {
    Cell cell = cells.get(i);
    // The scope will be null or empty if
    // there's nothing to replicate in that WALEdit
    if (!scopes.containsKey(cell.getFamily())
        || scopes.get(cell.getFamily()) == HConstants.REPLICATION_SCOPE_LOCAL) {
      cells.remove(i);
    }
  }
  if (cells.size() < size / 2) {
    cells.trimToSize();
  }
  return entry;
}