You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ns...@apache.org on 2011/09/27 04:42:01 UTC
svn commit: r1176177 [9/13] - in /hbase/branches/0.89: ./ bin/
bin/replication/ docs/ src/ src/assembly/ src/docs/
src/main/java/org/apache/hadoop/hbase/
src/main/java/org/apache/hadoop/hbase/avro/
src/main/java/org/apache/hadoop/hbase/avro/generated/ ...
Added: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java?rev=1176177&view=auto
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java (added)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/Replication.java Tue Sep 27 02:41:56 2011
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HServerInfo;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.regionserver.wal.HLog;
+import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
+import org.apache.hadoop.hbase.regionserver.wal.LogEntryVisitor;
+import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
+import org.apache.hadoop.hbase.replication.ReplicationZookeeperWrapper;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWrapper;
+
+import java.io.IOException;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Replication serves as an umbrella over the setup of replication and
+ * is used by HRS.
+ */
+public class Replication implements LogEntryVisitor {
+
+ private final boolean replication;
+ private final ReplicationSourceManager replicationManager;
+ private boolean replicationMaster;
+ private final AtomicBoolean replicating = new AtomicBoolean(true);
+ private final ReplicationZookeeperWrapper zkHelper;
+ private final Configuration conf;
+ private final AtomicBoolean stopRequested;
+ private ReplicationSink replicationSink;
+
+ /**
+ * Instantiate the replication management (if rep is enabled).
+ * @param conf conf to use
+ * @param hsi the info if this region server
+ * @param fs handle to the filesystem
+ * @param oldLogDir directory where logs are archived
+ * @param stopRequested boolean that tells us if we are shutting down
+ * @throws IOException
+ */
+ public Replication(Configuration conf, HServerInfo hsi,
+ FileSystem fs, Path oldLogDir,
+ AtomicBoolean stopRequested) throws IOException {
+ this.conf = conf;
+ this.stopRequested = stopRequested;
+ this.replication =
+ conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY, false);
+ if (replication) {
+ this.zkHelper = new ReplicationZookeeperWrapper(
+ ZooKeeperWrapper.getInstance(conf, hsi.getServerName()), conf,
+ this.replicating, hsi.getServerName());
+ this.replicationMaster = zkHelper.isReplicationMaster();
+ this.replicationManager = this.replicationMaster ?
+ new ReplicationSourceManager(zkHelper, conf, stopRequested,
+ fs, this.replicating, oldLogDir) : null;
+ } else {
+ replicationManager = null;
+ zkHelper = null;
+ }
+ }
+
+ /**
+ * Join with the replication threads
+ */
+ public void join() {
+ if (this.replication) {
+ if (this.replicationMaster) {
+ this.replicationManager.join();
+ }
+ this.zkHelper.deleteOwnRSZNode();
+ }
+ }
+
+ /**
+ * Carry on the list of log entries down to the sink
+ * @param entries list of entries to replicate
+ * @throws IOException
+ */
+ public void replicateLogEntries(HLog.Entry[] entries) throws IOException {
+ if (this.replication && !this.replicationMaster) {
+ this.replicationSink.replicateEntries(entries);
+ }
+ }
+
+ /**
+ * If replication is enabled and this cluster is a master,
+ * it starts
+ * @throws IOException
+ */
+ public void startReplicationServices() throws IOException {
+ if (this.replication) {
+ if (this.replicationMaster) {
+ this.replicationManager.init();
+ } else {
+ this.replicationSink =
+ new ReplicationSink(this.conf, this.stopRequested);
+ }
+ }
+ }
+
+ /**
+ * Get the replication sources manager
+ * @return the manager if replication is enabled, else returns false
+ */
+ public ReplicationSourceManager getReplicationManager() {
+ return replicationManager;
+ }
+
+ @Override
+ public void visitLogEntryBeforeWrite(HRegionInfo info, HLogKey logKey,
+ WALEdit logEdit) {
+ NavigableMap<byte[], Integer> scopes =
+ new TreeMap<byte[], Integer>(Bytes.BYTES_COMPARATOR);
+ byte[] family;
+ for (KeyValue kv : logEdit.getKeyValues()) {
+ family = kv.getFamily();
+ int scope = info.getTableDesc().getFamily(family).getScope();
+ if (scope != HConstants.REPLICATION_SCOPE_LOCAL &&
+ !scopes.containsKey(family)) {
+ scopes.put(family, scope);
+ }
+ }
+ if (!scopes.isEmpty()) {
+ logEdit.setScopes(scopes);
+ }
+ }
+
+ /**
+ * Add this class as a log entry visitor for HLog if replication is enabled
+ * @param hlog log that was add ourselves on
+ */
+ public void addLogEntryVisitor(HLog hlog) {
+ if (replication) {
+ hlog.addLogEntryVisitor(this);
+ }
+ }
+}
Added: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java?rev=1176177&view=auto
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java (added)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java Tue Sep 27 02:41:56 2011
@@ -0,0 +1,203 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.HTablePool;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.regionserver.wal.HLog;
+import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
+import org.apache.hadoop.hbase.util.Bytes;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * This class is responsible for replicating the edits coming
+ * from another cluster.
+ * <p/>
+ * This replication process is currently waiting for the edits to be applied
+ * before the method can return. This means that the replication of edits
+ * is synchronized (after reading from HLogs in ReplicationSource) and that a
+ * single region server cannot receive edits from two sources at the same time
+ * <p/>
+ * This class uses the native HBase client in order to replicate entries.
+ * <p/>
+ *
+ * TODO make this class more like ReplicationSource wrt log handling
+ */
+public class ReplicationSink {
+
+ private static final Log LOG = LogFactory.getLog(ReplicationSink.class);
+ // Name of the HDFS directory that contains the temporary rep logs
+ public static final String REPLICATION_LOG_DIR = ".replogs";
+ private final Configuration conf;
+ // Pool used to replicated
+ private final HTablePool pool;
+ // boolean coming from HRS to know when the process stops
+ private final AtomicBoolean stop;
+
+ /**
+ * Create a sink for replication
+ *
+ * @param conf conf object
+ * @param stopper boolean to tell this thread to stop
+ * @throws IOException thrown when HDFS goes bad or bad file name
+ */
+ public ReplicationSink(Configuration conf, AtomicBoolean stopper)
+ throws IOException {
+ this.conf = conf;
+ this.pool = new HTablePool(this.conf,
+ conf.getInt("replication.sink.htablepool.capacity", 10));
+ this.stop = stopper;
+ }
+
+ /**
+ * Replicate this array of entries directly into the local cluster
+ * using the native client.
+ *
+ * @param entries
+ * @throws IOException
+ */
+ public synchronized void replicateEntries(HLog.Entry[] entries)
+ throws IOException {
+ // Very simple optimization where we batch sequences of rows going
+ // to the same table.
+ try {
+ long totalReplicated = 0;
+ byte[] lastTable = HConstants.EMPTY_BYTE_ARRAY;
+ List<Put> puts = new ArrayList<Put>();
+ 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 {
+ // Switching table, flush
+ if (!Bytes.equals(lastTable, entry.getKey().getTablename())) {
+ put(lastTable, puts);
+ }
+ // With mini-batching, we need to expect multiple rows per edit
+ 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())) {
+ puts.add(put);
+ put = new Put(kv.getRow(), kv.getTimestamp());
+ }
+ put.add(kv.getFamily(), kv.getQualifier(), kv.getValue());
+ lastKey = kv.getRow();
+ }
+ puts.add(put);
+ lastTable = entry.getKey().getTablename();
+ }
+ totalReplicated++;
+ }
+ put(lastTable, puts);
+ LOG.info("Total replicated: " + totalReplicated);
+ } catch (IOException ex) {
+ if (ex.getCause() instanceof TableNotFoundException) {
+ LOG.warn("Losing edits because: ", ex);
+ } else {
+ // Should we log rejected edits in a file for replay?
+ LOG.error("Unable to accept edit because", ex);
+ this.stop.set(true);
+ throw ex;
+ }
+ } catch (RuntimeException re) {
+ if (re.getCause() instanceof TableNotFoundException) {
+ LOG.warn("Losing edits because: ", re);
+ } else {
+ this.stop.set(true);
+ throw re;
+ }
+ }
+ }
+
+ /**
+ * Do the puts and handle the pool
+ * @param tableName table to insert into
+ * @param puts list of puts
+ * @throws IOException
+ */
+ private void put(byte[] tableName, List<Put> puts) throws IOException {
+ if (puts.isEmpty()) {
+ return;
+ }
+ HTableInterface table = null;
+ try {
+ table = this.pool.getTable(tableName);
+ table.put(puts);
+ this.pool.putTable(table);
+ puts.clear();
+ } finally {
+ if (table != null) {
+ this.pool.putTable(table);
+ }
+ }
+ }
+
+ /**
+ * Do the delete and handle the pool
+ * @param tableName table to delete in
+ * @param delete the delete to use
+ * @throws IOException
+ */
+ private void delete(byte[] tableName, Delete delete) throws IOException {
+ HTableInterface table = null;
+ try {
+ table = this.pool.getTable(tableName);
+ table.delete(delete);
+ this.pool.putTable(table);
+ } finally {
+ if (table != null) {
+ this.pool.putTable(table);
+ }
+ }
+ }
+}
Added: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java?rev=1176177&view=auto
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java (added)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java Tue Sep 27 02:41:56 2011
@@ -0,0 +1,647 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HServerAddress;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.HConnection;
+import org.apache.hadoop.hbase.client.HConnectionManager;
+import org.apache.hadoop.hbase.ipc.HRegionInterface;
+import org.apache.hadoop.hbase.regionserver.wal.HLog;
+import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
+import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
+import org.apache.hadoop.hbase.replication.ReplicationZookeeperWrapper;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Threads;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.NavigableMap;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * Class that handles the source of a replication stream.
+ * Currently does not handle more than 1 slave
+ * For each slave cluster it selects a random number of peers
+ * using a replication ratio. For example, if replication ration = 0.1
+ * and slave cluster has 100 region servers, 10 will be selected.
+ * <p/>
+ * A stream is considered down when we cannot contact a region server on the
+ * peer cluster for more than 55 seconds by default.
+ * <p/>
+ *
+ */
+public class ReplicationSource extends Thread
+ implements ReplicationSourceInterface {
+
+ private static final Log LOG = LogFactory.getLog(ReplicationSource.class);
+ // Lock to manage when a HLog is changing path (archiving)
+ private final ReentrantLock pathLock = new ReentrantLock();
+ // Queue of logs to process
+ private PriorityBlockingQueue<Path> queue;
+ // container of entries to replicate
+ private HLog.Entry[] entriesArray;
+ private HConnection conn;
+ // Helper class for zookeeper
+ private ReplicationZookeeperWrapper zkHelper;
+ private Configuration conf;
+ // ratio of region servers to chose from a slave cluster
+ private float ratio;
+ private Random random;
+ // should we replicate or not?
+ private AtomicBoolean replicating;
+ // id of the peer cluster this source replicates to
+ private String peerClusterId;
+ // The manager of all sources to which we ping back our progress
+ private ReplicationSourceManager manager;
+ // Should we stop everything?
+ private AtomicBoolean stop;
+ // List of chosen sinks (region servers)
+ private List<HServerAddress> currentPeers;
+ // How long should we sleep for each retry
+ private long sleepForRetries;
+ // Max size in bytes of entriesArray
+ private long replicationQueueSizeCapacity;
+ // Max number of entries in entriesArray
+ private int replicationQueueNbCapacity;
+ // Our reader for the current log
+ private HLog.Reader reader;
+ // Current position in the log
+ private long position = 0;
+ // Path of the current log
+ private volatile Path currentPath;
+ private FileSystem fs;
+ // id of this cluster
+ private byte clusterId;
+ // total number of edits we replicated
+ private long totalReplicatedEdits = 0;
+ // The znode we currently play with
+ private String peerClusterZnode;
+ // Indicates if this queue is recovered (and will be deleted when depleted)
+ private boolean queueRecovered;
+ // Maximum number of retries before taking bold actions
+ private long maxRetriesMultiplier;
+ // Current number of entries that we need to replicate
+ private int currentNbEntries = 0;
+ // Indicates if this particular source is running
+ private volatile boolean running = true;
+
+ /**
+ * Instantiation method used by region servers
+ *
+ * @param conf configuration to use
+ * @param fs file system to use
+ * @param manager replication manager to ping to
+ * @param stopper the atomic boolean to use to stop the regionserver
+ * @param replicating the atomic boolean that starts/stops replication
+ * @param peerClusterZnode the name of our znode
+ * @throws IOException
+ */
+ public void init(final Configuration conf,
+ final FileSystem fs,
+ final ReplicationSourceManager manager,
+ final AtomicBoolean stopper,
+ final AtomicBoolean replicating,
+ final String peerClusterZnode)
+ throws IOException {
+ this.stop = stopper;
+ this.conf = conf;
+ this.replicationQueueSizeCapacity =
+ this.conf.getLong("replication.source.size.capacity", 1024*1024*64);
+ this.replicationQueueNbCapacity =
+ this.conf.getInt("replication.source.nb.capacity", 25000);
+ this.entriesArray = new HLog.Entry[this.replicationQueueNbCapacity];
+ for (int i = 0; i < this.replicationQueueNbCapacity; i++) {
+ this.entriesArray[i] = new HLog.Entry();
+ }
+ this.maxRetriesMultiplier =
+ this.conf.getLong("replication.source.maxretriesmultiplier", 10);
+ this.queue =
+ new PriorityBlockingQueue<Path>(
+ conf.getInt("hbase.regionserver.maxlogs", 32),
+ new LogsComparator());
+ this.conn = HConnectionManager.getConnection(conf);
+ this.zkHelper = manager.getRepZkWrapper();
+ this.ratio = this.conf.getFloat("replication.source.ratio", 0.1f);
+ this.currentPeers = new ArrayList<HServerAddress>();
+ this.random = new Random();
+ this.replicating = replicating;
+ this.manager = manager;
+ this.sleepForRetries =
+ this.conf.getLong("replication.source.sleepforretries", 1000);
+ this.fs = fs;
+ this.clusterId = Byte.valueOf(zkHelper.getClusterId());
+
+ // Finally look if this is a recovered queue
+ this.checkIfQueueRecovered(peerClusterZnode);
+ }
+
+ // The passed znode will be either the id of the peer cluster or
+ // the handling story of that queue in the form of id-startcode-*
+ private void checkIfQueueRecovered(String peerClusterZnode) {
+ String[] parts = peerClusterZnode.split("-");
+ this.queueRecovered = parts.length != 1;
+ this.peerClusterId = this.queueRecovered ?
+ parts[0] : peerClusterZnode;
+ this.peerClusterZnode = peerClusterZnode;
+ }
+
+ /**
+ * Select a number of peers at random using the ratio. Mininum 1.
+ */
+ private void chooseSinks() {
+ this.currentPeers.clear();
+ List<HServerAddress> addresses =
+ this.zkHelper.getPeersAddresses(peerClusterId);
+ Set<HServerAddress> setOfAddr = new HashSet<HServerAddress>();
+ int nbPeers = (int) (Math.ceil(addresses.size() * ratio));
+ LOG.info("Getting " + nbPeers +
+ " rs from peer cluster # " + peerClusterId);
+ for (int i = 0; i < nbPeers; i++) {
+ HServerAddress address;
+ // Make sure we get one address that we don't already have
+ do {
+ address = addresses.get(this.random.nextInt(addresses.size()));
+ } while (setOfAddr.contains(address));
+ LOG.info("Choosing peer " + address);
+ setOfAddr.add(address);
+ }
+ this.currentPeers.addAll(setOfAddr);
+ }
+
+ @Override
+ public void enqueueLog(Path log) {
+ this.queue.put(log);
+ }
+
+ @Override
+ public void logArchived(Path oldPath, Path newPath) {
+ // in sync with queue polling
+ this.pathLock.lock();
+ try {
+ if (oldPath.equals(this.currentPath)) {
+ this.currentPath = newPath;
+ LOG.debug("Current log moved, changing currentPath to " +newPath);
+ return;
+ }
+
+ boolean present = this.queue.remove(oldPath);
+ LOG.debug("old log was " + (present ?
+ "present, changing the queue" : "already processed"));
+ if (present) {
+ this.queue.add(newPath);
+ }
+ } finally {
+ this.pathLock.unlock();
+ }
+ }
+
+ @Override
+ public void run() {
+ connectToPeers();
+ // We were stopped while looping to connect to sinks, just abort
+ if (this.stop.get()) {
+ return;
+ }
+ // If this is recovered, the queue is already full and the first log
+ // normally has a position (unless the RS failed between 2 logs)
+ if (this.queueRecovered) {
+ this.position = this.zkHelper.getHLogRepPosition(
+ this.peerClusterZnode, this.queue.peek().getName());
+ }
+ int sleepMultiplier = 1;
+ // Loop until we close down
+ while (!stop.get() && this.running) {
+
+ // In sync with logArchived
+ this.pathLock.lock();
+ try {
+ // Get a new path
+ if (!getNextPath()) {
+ if (sleepForRetries("No log to process", sleepMultiplier)) {
+ sleepMultiplier++;
+ }
+ continue;
+ }
+ // Open a reader on it
+ if (!openReader(sleepMultiplier)) {
+ continue;
+ }
+ } finally {
+ this.pathLock.unlock();
+ }
+
+ // If we got a null reader but didn't continue, then sleep and continue
+ if (this.reader == null) {
+ if (sleepForRetries("Unable to open a reader", sleepMultiplier)) {
+ sleepMultiplier++;
+ }
+ continue;
+ }
+
+ boolean gotIOE = false; // TODO this is a hack for HDFS-1057
+ currentNbEntries = 0;
+ try {
+ if(readAllEntriesToReplicateOrNextFile()) {
+ continue;
+ }
+ } catch (IOException ioe) {
+ LOG.warn(peerClusterZnode + " Got: ", ioe);
+ gotIOE = true;
+ if (ioe.getCause() instanceof EOFException) {
+
+ boolean considerDumping = false;
+ if (this.queueRecovered) {
+ try {
+ FileStatus stat = this.fs.getFileStatus(this.currentPath);
+ if (stat.getLen() == 0) {
+ LOG.warn(peerClusterZnode + " Got EOF and the file was empty");
+ }
+ considerDumping = true;
+ } catch (IOException e) {
+ LOG.warn(peerClusterZnode + " Got while getting file size: ", e);
+ }
+ } else if (currentNbEntries != 0) {
+ LOG.warn(peerClusterZnode + " Got EOF while reading, " +
+ "looks like this file is broken? " + currentPath);
+ considerDumping = true;
+ currentNbEntries = 0;
+ }
+
+ if (considerDumping &&
+ sleepMultiplier == this.maxRetriesMultiplier &&
+ processEndOfFile()) {
+ continue;
+ }
+ }
+ } finally {
+ try {
+ // if current path is null, it means we processEndOfFile hence
+ if (this.currentPath != null && !gotIOE) {
+ this.position = this.reader.getPosition();
+ }
+ if (this.reader != null) {
+ this.reader.close();
+ }
+ } catch (IOException e) {
+ gotIOE = true;
+ LOG.warn("Unable to finalize the tailing of a file", e);
+ }
+ }
+
+ // If we didn't get anything to replicate, or if we hit a IOE,
+ // wait a bit and retry.
+ // But if we need to stop, don't bother sleeping
+ if (!stop.get() && (gotIOE || currentNbEntries == 0)) {
+ if (sleepForRetries("Nothing to replicate", sleepMultiplier)) {
+ sleepMultiplier++;
+ }
+ continue;
+ }
+ sleepMultiplier = 1;
+ shipEdits();
+
+ }
+ LOG.debug("Source exiting " + peerClusterId);
+ }
+
+ /**
+ * Read all the entries from the current log files and retain those
+ * that need to be replicated. Else, process the end of the current file.
+ * @return true if we got nothing and went to the next file, false if we got
+ * entries
+ * @throws IOException
+ */
+ protected boolean readAllEntriesToReplicateOrNextFile() throws IOException{
+ long seenEntries = 0;
+ if (this.position != 0) {
+ this.reader.seek(this.position);
+ }
+ HLog.Entry entry = this.reader.next(this.entriesArray[currentNbEntries]);
+ while (entry != null) {
+ WALEdit edit = entry.getEdit();
+ seenEntries++;
+ // Remove all KVs that should not be replicated
+ removeNonReplicableEdits(edit);
+ HLogKey logKey = entry.getKey();
+ // Don't replicate catalog entries, if the WALEdit wasn't
+ // containing anything to replicate and if we're currently not set to replicate
+ if (!(Bytes.equals(logKey.getTablename(), HConstants.ROOT_TABLE_NAME) ||
+ Bytes.equals(logKey.getTablename(), HConstants.META_TABLE_NAME)) &&
+ edit.size() != 0 && replicating.get()) {
+ logKey.setClusterId(this.clusterId);
+ currentNbEntries++;
+ }
+ // Stop if too many entries or too big
+ if ((this.reader.getPosition() - this.position)
+ >= this.replicationQueueSizeCapacity ||
+ currentNbEntries >= this.replicationQueueNbCapacity) {
+ break;
+ }
+ entry = this.reader.next(entriesArray[currentNbEntries]);
+ }
+ LOG.debug("currentNbEntries:" + currentNbEntries +
+ " and seenEntries:" + seenEntries +
+ " and size: " + (this.reader.getPosition() - this.position));
+ // If we didn't get anything and the queue has an object, it means we
+ // hit the end of the file for sure
+ return seenEntries == 0 && processEndOfFile();
+ }
+
+ private void connectToPeers() {
+ // Connect to peer cluster first, unless we have to stop
+ while (!this.stop.get() && this.currentPeers.size() == 0) {
+ try {
+ chooseSinks();
+ Thread.sleep(this.sleepForRetries);
+ } catch (InterruptedException e) {
+ LOG.error("Interrupted while trying to connect to sinks", e);
+ }
+ }
+ }
+
+ /**
+ * Poll for the next path
+ * @return true if a path was obtained, false if not
+ */
+ protected boolean getNextPath() {
+ try {
+ if (this.currentPath == null) {
+ this.currentPath = queue.poll(this.sleepForRetries, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted while reading edits", e);
+ }
+ return this.currentPath != null;
+ }
+
+ /**
+ * Open a reader on the current path
+ *
+ * @param sleepMultiplier by how many times the default sleeping time is augmented
+ * @return true if we should continue with that file, false if we are over with it
+ */
+ protected boolean openReader(int sleepMultiplier) {
+ try {
+ LOG.info("Opening log for replication " + this.currentPath.getName() + " at " + this.position);
+ this.reader = HLog.getReader(this.fs, this.currentPath, this.conf);
+ } catch (IOException ioe) {
+ this.reader = null;
+ LOG.warn(peerClusterZnode + " Got: ", ioe);
+ // TODO Need a better way to determinate if a file is really gone but
+ // TODO without scanning all logs dir
+ if (sleepMultiplier == this.maxRetriesMultiplier) {
+ LOG.warn("Waited too long for this file, considering dumping");
+ return !processEndOfFile();
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Do the sleeping logic
+ * @param msg Why we sleep
+ * @param sleepMultiplier by how many times the default sleeping time is augmented
+ * @return
+ */
+ protected boolean sleepForRetries(String msg, int sleepMultiplier) {
+ try {
+ LOG.debug(msg + ", sleeping " + sleepForRetries + " times " + sleepMultiplier);
+ Thread.sleep(this.sleepForRetries * sleepMultiplier);
+ } catch (InterruptedException e) {
+ LOG.debug("Interrupted while sleeping between retries");
+ }
+ return sleepMultiplier < maxRetriesMultiplier;
+ }
+
+ /**
+ * We only want KVs that are scoped other than local
+ * @param edit The KV to check for replication
+ */
+ protected void removeNonReplicableEdits(WALEdit edit) {
+ NavigableMap<byte[], Integer> scopes = edit.getScopes();
+ List<KeyValue> kvs = edit.getKeyValues();
+ for (int i = 0; i < edit.size(); i++) {
+ KeyValue kv = kvs.get(i);
+ // The scope will be null or empty if
+ // there's nothing to replicate in that WALEdit
+ if (scopes == null || !scopes.containsKey(kv.getFamily())) {
+ kvs.remove(i);
+ i--;
+ }
+ }
+ }
+
+ /**
+ * Do the shipping logic
+ */
+ protected void shipEdits() {
+ int sleepMultiplier = 1;
+ while (!stop.get()) {
+ try {
+ HRegionInterface rrs = getRS();
+ LOG.debug("Replicating " + currentNbEntries);
+ rrs.replicateLogEntries(Arrays.copyOf(this.entriesArray, currentNbEntries));
+ this.manager.logPositionAndCleanOldLogs(this.currentPath,
+ this.peerClusterZnode, this.position, queueRecovered);
+ this.totalReplicatedEdits += currentNbEntries;
+ LOG.debug("Replicated in total: " + this.totalReplicatedEdits);
+ break;
+
+ } catch (IOException ioe) {
+ LOG.warn("Unable to replicate because ", ioe);
+ try {
+ boolean down;
+ do {
+ down = isSlaveDown();
+ if (down) {
+ LOG.debug("The region server we tried to ping didn't answer, " +
+ "sleeping " + sleepForRetries + " times " + sleepMultiplier);
+ Thread.sleep(this.sleepForRetries * sleepMultiplier);
+ if (sleepMultiplier < maxRetriesMultiplier) {
+ sleepMultiplier++;
+ } else {
+ chooseSinks();
+ }
+ }
+ } while (!stop.get() && down);
+ } catch (InterruptedException e) {
+ LOG.debug("Interrupted while trying to contact the peer cluster");
+ }
+
+ }
+ }
+ }
+
+ /**
+ * If the queue isn't empty, switch to the next one
+ * Else if this is a recovered queue, it means we're done!
+ * Else we'll just continue to try reading the log file
+ * @return true if we're done with the current file, false if we should
+ * continue trying to read from it
+ */
+ protected boolean processEndOfFile() {
+ this.pathLock.lock();
+ try {
+ if (this.queue.size() != 0) {
+ this.currentPath = null;
+ this.position = 0;
+ return true;
+ } else if (this.queueRecovered) {
+ this.manager.closeRecoveredQueue(this);
+ this.abort();
+ return true;
+ }
+ } finally {
+ this.pathLock.unlock();
+ }
+ return false;
+ }
+
+ public void startup() {
+ String n = Thread.currentThread().getName();
+ Thread.UncaughtExceptionHandler handler =
+ new Thread.UncaughtExceptionHandler() {
+ public void uncaughtException(final Thread t, final Throwable e) {
+ LOG.fatal("Set stop flag in " + t.getName(), e);
+ abort();
+ }
+ };
+ Threads.setDaemonThreadRunning(
+ this, n + ".replicationSource," + clusterId, handler);
+ }
+
+ /**
+ * Hastily stop the replication, then wait for shutdown
+ */
+ private void abort() {
+ LOG.info("abort");
+ this.running = false;
+ terminate();
+ }
+
+ public void terminate() {
+ LOG.info("terminate");
+ Threads.shutdown(this, this.sleepForRetries);
+ }
+
+ /**
+ * Get a new region server at random from this peer
+ * @return
+ * @throws IOException
+ */
+ private HRegionInterface getRS() throws IOException {
+ if (this.currentPeers.size() == 0) {
+ throw new IOException(this.peerClusterZnode + " has 0 region servers");
+ }
+ HServerAddress address =
+ currentPeers.get(random.nextInt(this.currentPeers.size()));
+ return this.conn.getHRegionConnection(address);
+ }
+
+ /**
+ * Check if the slave is down by trying to establish a connection
+ * @return true if down, false if up
+ * @throws InterruptedException
+ */
+ public boolean isSlaveDown() throws InterruptedException {
+ final CountDownLatch latch = new CountDownLatch(1);
+ Thread pingThread = new Thread() {
+ public void run() {
+ try {
+ HRegionInterface rrs = getRS();
+ // Dummy call which should fail
+ rrs.getHServerInfo();
+ latch.countDown();
+ } catch (IOException ex) {
+ LOG.info("Slave cluster looks down: " + ex.getMessage());
+ }
+ }
+ };
+ pingThread.start();
+ // awaits returns true if countDown happened
+ boolean down = ! latch.await(this.sleepForRetries, TimeUnit.MILLISECONDS);
+ pingThread.interrupt();
+ return down;
+ }
+
+ /**
+ * Get the id that the source is replicating to
+ *
+ * @return peer cluster id
+ */
+ public String getPeerClusterZnode() {
+ return this.peerClusterZnode;
+ }
+
+ /**
+ * Get the path of the current HLog
+ * @return current hlog's path
+ */
+ public Path getCurrentPath() {
+ return this.currentPath;
+ }
+
+ /**
+ * Comparator used to compare logs together based on their start time
+ */
+ public static class LogsComparator implements Comparator<Path> {
+
+ @Override
+ public int compare(Path o1, Path o2) {
+ return Long.valueOf(getTS(o1)).compareTo(getTS(o2));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return true;
+ }
+
+ /**
+ * Split a path to get the start time
+ * For example: 10.20.20.171%3A60020.1277499063250
+ * @param p path to split
+ * @return start time
+ */
+ private long getTS(Path p) {
+ String[] parts = p.getName().split("\\.");
+ return Long.parseLong(parts[parts.length-1]);
+ }
+ }
+
+}
Added: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java?rev=1176177&view=auto
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java (added)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java Tue Sep 27 02:41:56 2011
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Interface that defines a replication source
+ */
+public interface ReplicationSourceInterface {
+
+ /**
+ * Initializer for the source
+ * @param conf the configuration to use
+ * @param fs the file system to use
+ * @param manager the manager to use
+ * @param stopper the stopper object for this region server
+ * @param replicating the status of the replication on this cluster
+ * @param peerClusterId the id of the peer cluster
+ * @throws IOException
+ */
+ public void init(final Configuration conf,
+ final FileSystem fs,
+ final ReplicationSourceManager manager,
+ final AtomicBoolean stopper,
+ final AtomicBoolean replicating,
+ final String peerClusterId) throws IOException;
+
+ /**
+ * Add a log to the list of logs to replicate
+ * @param log path to the log to replicate
+ */
+ public void enqueueLog(Path log);
+
+ /**
+ * Get the current log that's replicated
+ * @return the current log
+ */
+ public Path getCurrentPath();
+
+ /**
+ * Start the replication
+ */
+ public void startup();
+
+ /**
+ * End the replication
+ */
+ public void terminate();
+
+ /**
+ * Get the id that the source is replicating to
+ *
+ * @return peer cluster id
+ */
+ public String getPeerClusterZnode();
+
+ /**
+ * Notify this source that a log was archived
+ * @param oldPath old path of the log
+ * @param newPath new path of the log (archive)
+ */
+ public void logArchived(Path oldPath, Path newPath);
+}
Added: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java?rev=1176177&view=auto
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java (added)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java Tue Sep 27 02:41:56 2011
@@ -0,0 +1,353 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.regionserver.wal.LogActionsListener;
+import org.apache.hadoop.hbase.replication.ReplicationZookeeperWrapper;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * This class is responsible to manage all the replication
+ * sources. There are two classes of sources:
+ * <li> Normal sources are persistent and one per peer cluster</li>
+ * <li> Old sources are recovered from a failed region server and our
+ * only goal is to finish replicating the HLog queue it had up in ZK</li>
+ *
+ * When a region server dies, this class uses a watcher to get notified and it
+ * tries to grab a lock in order to transfer all the queues in a local
+ * old source.
+ */
+public class ReplicationSourceManager implements LogActionsListener {
+
+ private static final Log LOG =
+ LogFactory.getLog(ReplicationSourceManager.class);
+ // List of all the sources that read this RS's logs
+ private final List<ReplicationSourceInterface> sources;
+ // List of all the sources we got from died RSs
+ private final List<ReplicationSourceInterface> oldsources;
+ // Indicates if we are currently replicating
+ private final AtomicBoolean replicating;
+ // Helper for zookeeper
+ private final ReplicationZookeeperWrapper zkHelper;
+ // Indicates if the region server is closing
+ private final AtomicBoolean stopper;
+ // All logs we are currently trackign
+ private final SortedSet<String> hlogs;
+ private final Configuration conf;
+ private final FileSystem fs;
+ // The path to the latest log we saw, for new coming sources
+ private Path latestPath;
+ // List of all the other region servers in this cluster
+ private final List<String> otherRegionServers;
+ // Path to the hlog archive
+ private final Path oldLogDir;
+
+ /**
+ * Creates a replication manager and sets the watch on all the other
+ * registered region servers
+ * @param zkHelper the zk helper for replication
+ * @param conf the configuration to use
+ * @param stopper the stopper object for this region server
+ * @param fs the file system to use
+ * @param replicating the status of the replication on this cluster
+ * @param oldLogDir the directory where old logs are archived
+ */
+ public ReplicationSourceManager(final ReplicationZookeeperWrapper zkHelper,
+ final Configuration conf,
+ final AtomicBoolean stopper,
+ final FileSystem fs,
+ final AtomicBoolean replicating,
+ final Path oldLogDir) {
+ this.sources = new ArrayList<ReplicationSourceInterface>();
+ this.replicating = replicating;
+ this.zkHelper = zkHelper;
+ this.stopper = stopper;
+ this.hlogs = new TreeSet<String>();
+ this.oldsources = new ArrayList<ReplicationSourceInterface>();
+ this.conf = conf;
+ this.fs = fs;
+ this.oldLogDir = oldLogDir;
+ List<String> otherRSs =
+ this.zkHelper.getRegisteredRegionServers(new OtherRegionServerWatcher());
+ this.otherRegionServers = otherRSs == null ? new ArrayList<String>() : otherRSs;
+ }
+
+ /**
+ * Provide the id of the peer and a log key and this method will figure which
+ * hlog it belongs to and will log, for this region server, the current
+ * position. It will also clean old logs from the queue.
+ * @param log Path to the log currently being replicated from
+ * replication status in zookeeper. It will also delete older entries.
+ * @param id id of the peer cluster
+ * @param position current location in the log
+ * @param queueRecovered indicates if this queue comes from another region server
+ */
+ public void logPositionAndCleanOldLogs(Path log, String id, long position, boolean queueRecovered) {
+ String key = log.getName();
+ LOG.info("Going to report log #" + key + " for position " + position + " in " + log);
+ this.zkHelper.writeReplicationStatus(key.toString(), id, position);
+ synchronized (this.hlogs) {
+ if (!queueRecovered && this.hlogs.first() != key) {
+ SortedSet<String> hlogSet = this.hlogs.headSet(key);
+ LOG.info("Removing " + hlogSet.size() +
+ " logs in the list: " + hlogSet);
+ for (String hlog : hlogSet) {
+ this.zkHelper.removeLogFromList(hlog.toString(), id);
+ }
+ hlogSet.clear();
+ }
+ }
+ }
+
+ /**
+ * Adds a normal source per registered peer cluster and tries to process all
+ * old region server hlog queues
+ */
+ public void init() throws IOException {
+ for (String id : this.zkHelper.getPeerClusters().keySet()) {
+ ReplicationSourceInterface src = addSource(id);
+ src.startup();
+ }
+ List<String> currentReplicators = this.zkHelper.getListOfReplicators(null);
+ synchronized (otherRegionServers) {
+ LOG.info("Current list of replicators: " + currentReplicators
+ + " other RSs: " + otherRegionServers);
+ }
+ // Look if there's anything to process after a restart
+ for (String rs : currentReplicators) {
+ synchronized (otherRegionServers) {
+ if (!this.otherRegionServers.contains(rs)) {
+ transferQueues(rs);
+ }
+ }
+ }
+ }
+
+ /**
+ * Add a new normal source to this region server
+ * @param id the id of the peer cluster
+ * @return the created source
+ * @throws IOException
+ */
+ public ReplicationSourceInterface addSource(String id) throws IOException {
+ ReplicationSourceInterface src =
+ getReplicationSource(this.conf, this.fs, this, stopper, replicating, id);
+ this.sources.add(src);
+ synchronized (this.hlogs) {
+ if (this.hlogs.size() > 0) {
+ this.zkHelper.addLogToList(this.hlogs.first(),
+ this.sources.get(0).getPeerClusterZnode());
+ src.enqueueLog(this.latestPath);
+ }
+ }
+ return src;
+ }
+
+ /**
+ * Terminate the replication on this region server
+ */
+ public void join() {
+ if (this.sources.size() == 0) {
+ this.zkHelper.deleteOwnRSZNode();
+ }
+ for (ReplicationSourceInterface source : this.sources) {
+ source.terminate();
+ }
+ }
+
+ /**
+ * Get a copy of the hlogs of the first source on this rs
+ * @return a sorted set of hlog names
+ */
+ protected SortedSet<String> getHLogs() {
+ return new TreeSet(this.hlogs);
+ }
+
+ /**
+ * Get a list of all the normal sources of this rs
+ * @return lis of all sources
+ */
+ public List<ReplicationSourceInterface> getSources() {
+ return this.sources;
+ }
+
+ @Override
+ public void logRolled(Path newLog) {
+ if (this.sources.size() > 0) {
+ this.zkHelper.addLogToList(newLog.getName(),
+ this.sources.get(0).getPeerClusterZnode());
+ }
+ synchronized (this.hlogs) {
+ this.hlogs.add(newLog.getName());
+ }
+ this.latestPath = newLog;
+ for (ReplicationSourceInterface source : this.sources) {
+ source.enqueueLog(newLog);
+ }
+ }
+
+ @Override
+ public void logArchived(Path oldPath, Path newPath) {
+ for (ReplicationSourceInterface source : this.sources) {
+ source.logArchived(oldPath, newPath);
+ }
+ }
+
+ /**
+ * Get the ZK help of this manager
+ * @return the helper
+ */
+ public ReplicationZookeeperWrapper getRepZkWrapper() {
+ return zkHelper;
+ }
+
+ /**
+ * Factory method to create a replication source
+ * @param conf the configuration to use
+ * @param fs the file system to use
+ * @param manager the manager to use
+ * @param stopper the stopper object for this region server
+ * @param replicating the status of the replication on this cluster
+ * @param peerClusterId the id of the peer cluster
+ * @return the created source
+ * @throws IOException
+ */
+ public ReplicationSourceInterface getReplicationSource(
+ final Configuration conf,
+ final FileSystem fs,
+ final ReplicationSourceManager manager,
+ final AtomicBoolean stopper,
+ final AtomicBoolean replicating,
+ final String peerClusterId) throws IOException {
+ ReplicationSourceInterface src;
+ try {
+ Class c = Class.forName(conf.get("replication.replicationsource.implementation",
+ ReplicationSource.class.getCanonicalName()));
+ src = (ReplicationSourceInterface) c.newInstance();
+ } catch (Exception e) {
+ LOG.warn("Passed replication source implemention throws errors, " +
+ "defaulting to ReplicationSource", e);
+ src = new ReplicationSource();
+
+ }
+ src.init(conf, fs, manager, stopper, replicating, peerClusterId);
+ return src;
+ }
+
+ /**
+ * Transfer all the queues of the specified to this region server.
+ * First it tries to grab a lock and if it works it will move the
+ * znodes and finally will delete the old znodes.
+ *
+ * It creates one old source for any type of source of the old rs.
+ * @param rsZnode
+ */
+ public void transferQueues(String rsZnode) {
+ // We try to lock that rs' queue directory
+ if (this.stopper.get()) {
+ LOG.info("Not transferring queue since we are shutting down");
+ return;
+ }
+ if (!this.zkHelper.lockOtherRS(rsZnode)) {
+ return;
+ }
+ LOG.info("Moving " + rsZnode + "'s hlogs to my queue");
+ SortedMap<String, SortedSet<String>> newQueues =
+ this.zkHelper.copyQueuesFromRS(rsZnode);
+ if (newQueues == null || newQueues.size() == 0) {
+ return;
+ }
+ this.zkHelper.deleteRsQueues(rsZnode);
+
+ for (Map.Entry<String, SortedSet<String>> entry : newQueues.entrySet()) {
+ String peerId = entry.getKey();
+ try {
+ ReplicationSourceInterface src = getReplicationSource(this.conf,
+ this.fs, this, this.stopper, this.replicating, peerId);
+ this.oldsources.add(src);
+ for (String hlog : entry.getValue()) {
+ src.enqueueLog(new Path(this.oldLogDir, hlog));
+ }
+ src.startup();
+ } catch (IOException e) {
+ // TODO manage it
+ LOG.error("Failed creating a source", e);
+ }
+ }
+ }
+
+ /**
+ * Clear the references to the specified old source
+ * @param src source to clear
+ */
+ public void closeRecoveredQueue(ReplicationSourceInterface src) {
+ LOG.info("Done with the recovered queue " + src.getPeerClusterZnode());
+ this.oldsources.remove(src);
+ this.zkHelper.deleteSource(src.getPeerClusterZnode());
+ }
+
+ /**
+ * Watcher used to be notified of the other region server's death
+ * in the local cluster. It initiates the process to transfer the queues
+ * if it is able to grab the lock.
+ */
+ public class OtherRegionServerWatcher implements Watcher {
+ @Override
+ public void process(WatchedEvent watchedEvent) {
+ LOG.info(" event " + watchedEvent);
+ if (watchedEvent.getType().equals(Event.KeeperState.Expired) ||
+ watchedEvent.getType().equals(Event.KeeperState.Disconnected)) {
+ return;
+ }
+
+ List<String> newRsList = (zkHelper.getRegisteredRegionServers(this));
+ if (newRsList == null) {
+ return;
+ } else {
+ synchronized (otherRegionServers) {
+ otherRegionServers.clear();
+ otherRegionServers.addAll(newRsList);
+ }
+ }
+ if (watchedEvent.getType().equals(Event.EventType.NodeDeleted)) {
+ LOG.info(watchedEvent.getPath() + " znode expired, trying to lock it");
+ String[] rsZnodeParts = watchedEvent.getPath().split("/");
+ transferQueues(rsZnodeParts[rsZnodeParts.length-1]);
+ }
+ }
+ }
+
+}
Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/BloomFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/BloomFilter.java?rev=1176177&r1=1176176&r2=1176177&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/BloomFilter.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/BloomFilter.java Tue Sep 27 02:41:56 2011
@@ -1,5 +1,5 @@
-/**
- * Copyright 2009 The Apache Software Foundation
+/*
+ * Copyright 2010 The Apache Software Foundation
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,38 +19,37 @@
*/
package org.apache.hadoop.hbase.util;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-
import org.apache.hadoop.io.Writable;
+import java.nio.ByteBuffer;
+
/**
* Defines the general behavior of a bloom filter.
* <p>
- * The Bloom filter is a data structure that was introduced in 1970 and that has been adopted by
+ * The Bloom filter is a data structure that was introduced in 1970 and that has been adopted by
* the networking research community in the past decade thanks to the bandwidth efficiencies that it
- * offers for the transmission of set membership information between networked hosts. A sender encodes
- * the information into a bit vector, the Bloom filter, that is more compact than a conventional
- * representation. Computation and space costs for construction are linear in the number of elements.
- * The receiver uses the filter to test whether various elements are members of the set. Though the
- * filter will occasionally return a false positive, it will never return a false negative. When creating
- * the filter, the sender can choose its desired point in a trade-off between the false positive rate and the size.
- *
+ * offers for the transmission of set membership information between networked hosts. A sender encodes
+ * the information into a bit vector, the Bloom filter, that is more compact than a conventional
+ * representation. Computation and space costs for construction are linear in the number of elements.
+ * The receiver uses the filter to test whether various elements are members of the set. Though the
+ * filter will occasionally return a false positive, it will never return a false negative. When creating
+ * the filter, the sender can choose its desired point in a trade-off between the false positive rate and the size.
+ *
* <p>
* Originally created by
* <a href="http://www.one-lab.org">European Commission One-Lab Project 034819</a>.
- *
+ *
* <p>
* It must be extended in order to define the real behavior.
*/
public interface BloomFilter {
- /**
- * Allocate memory for the bloom filter data. Note that bloom data isn't
- * allocated by default because it can grow large & reads would be better
+ /**
+ * Allocate memory for the bloom filter data. Note that bloom data isn't
+ * allocated by default because it can grow large & reads would be better
* managed by the LRU cache.
*/
void allocBloom();
-
+
/**
* Add the specified binary to the bloom filter.
*
@@ -86,7 +85,7 @@ public interface BloomFilter {
* @return true if matched by bloom, false if not
*/
boolean contains(byte [] buf, int offset, int length, ByteBuffer bloom);
-
+
/**
* @return The number of keys added to the bloom
*/
@@ -97,16 +96,16 @@ public interface BloomFilter {
* to maintain the desired error rate
*/
public int getMaxKeys();
-
+
/**
- * Size of the bloom, in bytes
+ * @return Size of the bloom, in bytes
*/
public int getByteSize();
-
+
/**
- * Finalize the bloom before writing metadata & data to disk
+ * Compact the bloom before writing metadata & data to disk
*/
- void finalize();
+ void compactBloom();
/**
* Get a writable interface into bloom filter meta data.
Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/ByteBloomFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/ByteBloomFilter.java?rev=1176177&r1=1176176&r2=1176177&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/ByteBloomFilter.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/ByteBloomFilter.java Tue Sep 27 02:41:56 2011
@@ -1,35 +1,6 @@
-/**
+/*
+ * Copyright 2010 The Apache Software Foundation
*
- * Copyright (c) 2005, European Commission project OneLab under contract 034819 (http://www.one-lab.org)
- * All rights reserved.
- * Redistribution and use in source and binary forms, with or
- * without modification, are permitted provided that the following
- * conditions are met:
- * - Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * - Redistributions in binary form must reproduce the above copyright
- * notice, this list of conditions and the following disclaimer in
- * the documentation and/or other materials provided with the distribution.
- * - Neither the name of the University Catholique de Louvain - UCL
- * nor the names of its contributors may be used to endorse or
- * promote products derived from this software without specific prior
- * written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
- * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
- * COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
- * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
- * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
- * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
- * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
- * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
- * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
- * POSSIBILITY OF SUCH DAMAGE.
- */
-
-/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -48,41 +19,38 @@
*/
package org.apache.hadoop.hbase.util;
-
-import java.io.DataOutput;
+
+import org.apache.hadoop.io.Writable;
+
import java.io.DataInput;
+import java.io.DataOutput;
import java.io.IOException;
-import java.lang.Math;
import java.nio.ByteBuffer;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.util.bloom.Filter;
-
/**
* Implements a <i>Bloom filter</i>, as defined by Bloom in 1970.
* <p>
- * The Bloom filter is a data structure that was introduced in 1970 and that has been adopted by
+ * The Bloom filter is a data structure that was introduced in 1970 and that has been adopted by
* the networking research community in the past decade thanks to the bandwidth efficiencies that it
- * offers for the transmission of set membership information between networked hosts. A sender encodes
- * the information into a bit vector, the Bloom filter, that is more compact than a conventional
- * representation. Computation and space costs for construction are linear in the number of elements.
- * The receiver uses the filter to test whether various elements are members of the set. Though the
- * filter will occasionally return a false positive, it will never return a false negative. When creating
- * the filter, the sender can choose its desired point in a trade-off between the false positive rate and the size.
- *
+ * offers for the transmission of set membership information between networked hosts. A sender encodes
+ * the information into a bit vector, the Bloom filter, that is more compact than a conventional
+ * representation. Computation and space costs for construction are linear in the number of elements.
+ * The receiver uses the filter to test whether various elements are members of the set. Though the
+ * filter will occasionally return a false positive, it will never return a false negative. When creating
+ * the filter, the sender can choose its desired point in a trade-off between the false positive rate and the size.
+ *
* <p>
- * Originally created by
+ * Originally inspired by
* <a href="http://www.one-lab.org">European Commission One-Lab Project 034819</a>.
- *
+ *
* @see BloomFilter The general behavior of a filter
- *
+ *
* @see <a href="http://portal.acm.org/citation.cfm?id=362692&dl=ACM&coll=portal">Space/Time Trade-Offs in Hash Coding with Allowable Errors</a>
*/
public class ByteBloomFilter implements BloomFilter {
/** Current file format version */
public static final int VERSION = 1;
-
+
/** Bytes (B) in the array */
protected int byteSize;
/** Number of hash functions */
@@ -97,7 +65,7 @@ public class ByteBloomFilter implements
protected int maxKeys;
/** Bloom bits */
protected ByteBuffer bloom;
-
+
/** Bit-value lookup array to prevent doing the same work over and over */
private static final byte [] bitvals = {
(byte) 0x01,
@@ -115,11 +83,11 @@ public class ByteBloomFilter implements
* @param meta stored bloom meta data
* @throws IllegalArgumentException meta data is invalid
*/
- public ByteBloomFilter(ByteBuffer meta)
- throws IllegalArgumentException {
+ public ByteBloomFilter(ByteBuffer meta)
+ throws IllegalArgumentException {
int version = meta.getInt();
if (version != VERSION) throw new IllegalArgumentException("Bad version");
-
+
this.byteSize = meta.getInt();
this.hashCount = meta.getInt();
this.hashType = meta.getInt();
@@ -136,30 +104,30 @@ public class ByteBloomFilter implements
* @param maxKeys Maximum expected number of keys that will be stored in this bloom
* @param errorRate Desired false positive error rate. Lower rate = more storage required
* @param hashType Type of hash function to use
- * @param foldFactor When finished adding entries, you may be able to 'fold'
- * this bloom to save space. Tradeoff potentially excess bytes in bloom for
+ * @param foldFactor When finished adding entries, you may be able to 'fold'
+ * this bloom to save space. Tradeoff potentially excess bytes in bloom for
* ability to fold if keyCount is exponentially greater than maxKeys.
* @throws IllegalArgumentException
*/
public ByteBloomFilter(int maxKeys, float errorRate, int hashType, int foldFactor)
- throws IllegalArgumentException {
- /*
- * Bloom filters are very sensitive to the number of elements inserted
- * into them. For HBase, the number of entries depends on the size of the
- * data stored in the column. Currently the default region size is 256MB,
- * so entry count ~= 256MB / (average value size for column). Despite
- * this rule of thumb, there is no efficient way to calculate the entry
- * count after compactions. Therefore, it is often easier to use a
+ throws IllegalArgumentException {
+ /*
+ * Bloom filters are very sensitive to the number of elements inserted
+ * into them. For HBase, the number of entries depends on the size of the
+ * data stored in the column. Currently the default region size is 256MB,
+ * so entry count ~= 256MB / (average value size for column). Despite
+ * this rule of thumb, there is no efficient way to calculate the entry
+ * count after compactions. Therefore, it is often easier to use a
* dynamic bloom filter that will add extra space instead of allowing the
* error rate to grow.
- *
+ *
* ( http://www.eecs.harvard.edu/~michaelm/NEWWORK/postscripts/BloomFilterSurvey.pdf )
*
* m denotes the number of bits in the Bloom filter (bitSize)
* n denotes the number of elements inserted into the Bloom filter (maxKeys)
* k represents the number of hash functions used (nbHash)
* e represents the desired false positive rate for the bloom (err)
- *
+ *
* If we fix the error rate (e) and know the number of entries, then
* the optimal bloom size m = -(n * ln(err) / (ln(2)^2)
* ~= n * ln(err) / ln(0.6185)
@@ -196,12 +164,12 @@ public class ByteBloomFilter implements
this.bloom = ByteBuffer.allocate(this.byteSize);
assert this.bloom.hasArray();
}
-
+
void sanityCheck() throws IllegalArgumentException {
if(this.byteSize <= 0) {
throw new IllegalArgumentException("maxValue must be > 0");
}
-
+
if(this.hashCount <= 0) {
throw new IllegalArgumentException("Hash function count must be > 0");
}
@@ -209,12 +177,12 @@ public class ByteBloomFilter implements
if (this.hash == null) {
throw new IllegalArgumentException("hashType must be known");
}
-
+
if (this.keyCount < 0) {
throw new IllegalArgumentException("must have positive keyCount");
}
}
-
+
void bloomCheck(ByteBuffer bloom) throws IllegalArgumentException {
if (this.byteSize != bloom.limit()) {
throw new IllegalArgumentException(
@@ -243,14 +211,14 @@ public class ByteBloomFilter implements
++this.keyCount;
}
-
+
/**
* Should only be used in tests when writing a bloom filter.
*/
boolean contains(byte [] buf) {
return contains(buf, 0, buf.length, this.bloom);
}
-
+
/**
* Should only be used in tests when writing a bloom filter.
*/
@@ -264,7 +232,7 @@ public class ByteBloomFilter implements
}
@Override
- public boolean contains(byte [] buf, int offset, int length,
+ public boolean contains(byte [] buf, int offset, int length,
ByteBuffer theBloom) {
if(theBloom.limit() != this.byteSize) {
@@ -282,11 +250,11 @@ public class ByteBloomFilter implements
}
return true;
}
-
+
//---------------------------------------------------------------------------
/** Private helpers */
-
- /**
+
+ /**
* Set the bit at the specified index to 1.
*
* @param pos index of bit
@@ -298,7 +266,7 @@ public class ByteBloomFilter implements
curByte |= bitvals[bitPos];
bloom.put(bytePos, curByte);
}
-
+
/**
* Check if bit at specified index is 1.
*
@@ -312,37 +280,37 @@ public class ByteBloomFilter implements
curByte &= bitvals[bitPos];
return (curByte != 0);
}
-
+
@Override
public int getKeyCount() {
return this.keyCount;
}
-
+
@Override
public int getMaxKeys() {
return this.maxKeys;
}
-
+
@Override
public int getByteSize() {
return this.byteSize;
}
@Override
- public void finalize() {
+ public void compactBloom() {
// see if the actual size is exponentially smaller than expected.
if (this.keyCount > 0 && this.bloom.hasArray()) {
int pieces = 1;
int newByteSize = this.byteSize;
int newMaxKeys = this.maxKeys;
-
+
// while exponentially smaller & folding is lossless
while ( (newByteSize & 1) == 0 && newMaxKeys > (this.keyCount<<1) ) {
pieces <<= 1;
newByteSize >>= 1;
newMaxKeys >>= 1;
}
-
+
// if we should fold these into pieces
if (pieces > 1) {
byte[] array = this.bloom.array();
@@ -351,7 +319,7 @@ public class ByteBloomFilter implements
int off = end;
for(int p = 1; p < pieces; ++p) {
for(int pos = start; pos < end; ++pos) {
- array[pos] |= array[off++];
+ array[pos] |= array[off++];
}
}
// folding done, only use a subset of this array
@@ -388,9 +356,9 @@ public class ByteBloomFilter implements
public Writable getDataWriter() {
return new DataWriter();
}
-
+
private class MetaWriter implements Writable {
- protected MetaWriter() {}
+ protected MetaWriter() {}
@Override
public void readFields(DataInput arg0) throws IOException {
throw new IOException("Cant read with this class.");
Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/DynamicByteBloomFilter.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/DynamicByteBloomFilter.java?rev=1176177&r1=1176176&r2=1176177&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/DynamicByteBloomFilter.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/DynamicByteBloomFilter.java Tue Sep 27 02:41:56 2011
@@ -1,35 +1,6 @@
-/**
+/*
+ * Copyright 2010 The Apache Software Foundation
*
- * Copyright (c) 2005, European Commission project OneLab under contract 034819 (http://www.one-lab.org)
- * All rights reserved.
- * Redistribution and use in source and binary forms, with or
- * without modification, are permitted provided that the following
- * conditions are met:
- * - Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * - Redistributions in binary form must reproduce the above copyright
- * notice, this list of conditions and the following disclaimer in
- * the documentation and/or other materials provided with the distribution.
- * - Neither the name of the University Catholique de Louvain - UCL
- * nor the names of its contributors may be used to endorse or
- * promote products derived from this software without specific prior
- * written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
- * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
- * COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
- * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
- * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
- * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
- * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
- * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
- * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
- * POSSIBILITY OF SUCH DAMAGE.
- */
-
-/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -49,30 +20,30 @@
package org.apache.hadoop.hbase.util;
+import org.apache.hadoop.io.Writable;
+
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.nio.ByteBuffer;
-import org.apache.hadoop.io.Writable;
-
/**
* Implements a <i>dynamic Bloom filter</i>, as defined in the INFOCOM 2006 paper.
* <p>
* A dynamic Bloom filter (DBF) makes use of a <code>s * m</code> bit matrix but
- * each of the <code>s</code> rows is a standard Bloom filter. The creation
+ * each of the <code>s</code> rows is a standard Bloom filter. The creation
* process of a DBF is iterative. At the start, the DBF is a <code>1 * m</code>
* bit matrix, i.e., it is composed of a single standard Bloom filter.
- * It assumes that <code>n<sub>r</sub></code> elements are recorded in the
+ * It assumes that <code>n<sub>r</sub></code> elements are recorded in the
* initial bit vector, where <code>n<sub>r</sub> <= n</code> (<code>n</code> is
- * the cardinality of the set <code>A</code> to record in the filter).
+ * the cardinality of the set <code>A</code> to record in the filter).
* <p>
* As the size of <code>A</code> grows during the execution of the application,
* several keys must be inserted in the DBF. When inserting a key into the DBF,
* one must first get an active Bloom filter in the matrix. A Bloom filter is
- * active when the number of recorded keys, <code>n<sub>r</sub></code>, is
+ * active when the number of recorded keys, <code>n<sub>r</sub></code>, is
* strictly less than the current cardinality of <code>A</code>, <code>n</code>.
- * If an active Bloom filter is found, the key is inserted and
+ * If an active Bloom filter is found, the key is inserted and
* <code>n<sub>r</sub></code> is incremented by one. On the other hand, if there
* is no active Bloom filter, a new one is created (i.e., a new row is added to
* the matrix) according to the current size of <code>A</code> and the element
@@ -84,7 +55,7 @@ import org.apache.hadoop.io.Writable;
* <a href="http://www.one-lab.org">European Commission One-Lab Project 034819</a>.
*
* @see BloomFilter A Bloom filter
- *
+ *
* @see <a href="http://www.cse.fau.edu/~jie/research/publications/Publication_files/infocom2006.pdf">Theory and Network Applications of Dynamic Bloom Filters</a>
*/
public class DynamicByteBloomFilter implements BloomFilter {
@@ -108,8 +79,7 @@ public class DynamicByteBloomFilter impl
* @param meta stored bloom meta data
* @throws IllegalArgumentException meta data is invalid
*/
- public DynamicByteBloomFilter(ByteBuffer meta)
- throws IllegalArgumentException {
+ public DynamicByteBloomFilter(ByteBuffer meta) throws IllegalArgumentException {
int version = meta.getInt();
if (version != VERSION) throw new IllegalArgumentException("Bad version");
@@ -118,7 +88,7 @@ public class DynamicByteBloomFilter impl
this.hashType = meta.getInt();
this.readMatrixSize = meta.getInt();
this.curKeys = meta.getInt();
-
+
readSanityCheck();
this.matrix = new ByteBloomFilter[1];
@@ -126,12 +96,9 @@ public class DynamicByteBloomFilter impl
}
/**
- * Normal write constructor. Note that this doesn't allocate bloom data by
+ * Normal write constructor. Note that this doesn't allocate bloom data by
* default. Instead, call allocBloom() before adding entries.
- * @param bitSize The vector size of <i>this</i> filter.
- * @param functionCount The number of hash function to consider.
- * @param hashType type of the hashing function (see
- * {@link org.apache.hadoop.util.hash.Hash}).
+ * @param hashType type of the hashing function (see {@link org.apache.hadoop.util.hash.Hash}).
* @param keyInterval Maximum number of keys to record per Bloom filter row.
* @throws IllegalArgumentException The input parameters were invalid
*/
@@ -141,7 +108,7 @@ public class DynamicByteBloomFilter impl
this.errorRate = errorRate;
this.hashType = hashType;
this.curKeys = 0;
-
+
if(keyInterval <= 0) {
throw new IllegalArgumentException("keyCount must be > 0");
}
@@ -164,7 +131,7 @@ public class DynamicByteBloomFilter impl
throw new IllegalArgumentException("matrix size must be known");
}
}
-
+
@Override
public void add(byte []buf, int offset, int len) {
BloomFilter bf = getCurBloom();
@@ -207,17 +174,17 @@ public class DynamicByteBloomFilter impl
public boolean contains(byte [] buf, ByteBuffer theBloom) {
return contains(buf, 0, buf.length, theBloom);
}
-
+
@Override
- public boolean contains(byte[] buf, int offset, int length,
+ public boolean contains(byte[] buf, int offset, int length,
ByteBuffer theBloom) {
if(offset + length > buf.length) {
return false;
}
-
+
// current version assumes uniform size
- int bytesPerBloom = this.matrix[0].getByteSize();
-
+ int bytesPerBloom = this.matrix[0].getByteSize();
+
if(theBloom.limit() != bytesPerBloom * readMatrixSize) {
throw new IllegalArgumentException("Bloom does not match expected size");
}
@@ -233,7 +200,7 @@ public class DynamicByteBloomFilter impl
return true;
}
}
-
+
// matched no bloom filters
return false;
}
@@ -251,14 +218,14 @@ public class DynamicByteBloomFilter impl
public int getMaxKeys() {
return bloomCount() * this.keyInterval;
}
-
+
@Override
public int getByteSize() {
return bloomCount() * this.matrix[0].getByteSize();
}
@Override
- public void finalize() {
+ public void compactBloom() {
}
/**
@@ -298,9 +265,9 @@ public class DynamicByteBloomFilter impl
public Writable getDataWriter() {
return new DataWriter();
}
-
+
private class MetaWriter implements Writable {
- protected MetaWriter() {}
+ protected MetaWriter() {}
@Override
public void readFields(DataInput arg0) throws IOException {
throw new IOException("Cant read with this class.");
Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java?rev=1176177&r1=1176176&r2=1176177&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/util/FSUtils.java Tue Sep 27 02:41:56 2011
@@ -601,6 +601,7 @@ public class FSUtils {
public static void recoverFileLease(final FileSystem fs, final Path p, Configuration conf)
throws IOException{
if (!isAppendSupported(conf)) {
+ LOG.warn("Running on HDFS without append enabled may result in data loss");
return;
}
// lease recovery not needed for local file system case.
Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/HQuorumPeer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/HQuorumPeer.java?rev=1176177&r1=1176176&r2=1176177&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/HQuorumPeer.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/HQuorumPeer.java Tue Sep 27 02:41:56 2011
@@ -51,7 +51,7 @@ import java.util.Properties;
* control over the process. Currently, this class allows us to parse the
* zoo.cfg and inject variables from HBase's site.xml configuration in.
*/
-public class HQuorumPeer implements HConstants {
+public class HQuorumPeer {
private static final Log LOG = LogFactory.getLog(HQuorumPeer.class);
private static final String VARIABLE_START = "${";
@@ -171,12 +171,13 @@ public class HQuorumPeer implements HCon
// First check if there is a zoo.cfg in the CLASSPATH. If so, simply read
// it and grab its configuration properties.
ClassLoader cl = HQuorumPeer.class.getClassLoader();
- InputStream inputStream = cl.getResourceAsStream(ZOOKEEPER_CONFIG_NAME);
+ final InputStream inputStream =
+ cl.getResourceAsStream(HConstants.ZOOKEEPER_CONFIG_NAME);
if (inputStream != null) {
try {
return parseZooCfg(conf, inputStream);
} catch (IOException e) {
- LOG.warn("Cannot read " + ZOOKEEPER_CONFIG_NAME +
+ LOG.warn("Cannot read " + HConstants.ZOOKEEPER_CONFIG_NAME +
", loading from XML files", e);
}
}
@@ -200,14 +201,16 @@ public class HQuorumPeer implements HCon
// If clientPort is not set, assign the default
if (zkProperties.getProperty(ZK_CLIENT_PORT_KEY) == null) {
- zkProperties.put(ZK_CLIENT_PORT_KEY, DEFAULT_ZOOKEPER_CLIENT_PORT);
+ zkProperties.put(ZK_CLIENT_PORT_KEY,
+ HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT);
}
// Create the server.X properties.
int peerPort = conf.getInt("hbase.zookeeper.peerport", 2888);
int leaderPort = conf.getInt("hbase.zookeeper.leaderport", 3888);
- String[] serverHosts = conf.getStrings(ZOOKEEPER_QUORUM, "localhost");
+ final String[] serverHosts = conf.getStrings(HConstants.ZOOKEEPER_QUORUM,
+ "localhost");
for (int i = 0; i < serverHosts.length; ++i) {
String serverHost = serverHosts[i];
String address = serverHost + ":" + peerPort + ":" + leaderPort;
@@ -219,6 +222,68 @@ public class HQuorumPeer implements HCon
}
/**
+ * Return the ZK Quorum servers string given zk properties returned by
+ * makeZKProps
+ * @param properties
+ * @return
+ */
+ public static String getZKQuorumServersString(Properties properties) {
+ String clientPort = null;
+ List<String> servers = new ArrayList<String>();
+
+ // The clientPort option may come after the server.X hosts, so we need to
+ // grab everything and then create the final host:port comma separated list.
+ boolean anyValid = false;
+ for (Entry<Object,Object> property : properties.entrySet()) {
+ String key = property.getKey().toString().trim();
+ String value = property.getValue().toString().trim();
+ if (key.equals("clientPort")) {
+ clientPort = value;
+ }
+ else if (key.startsWith("server.")) {
+ String host = value.substring(0, value.indexOf(':'));
+ servers.add(host);
+ try {
+ //noinspection ResultOfMethodCallIgnored
+ InetAddress.getByName(host);
+ anyValid = true;
+ } catch (UnknownHostException e) {
+ LOG.warn(StringUtils.stringifyException(e));
+ }
+ }
+ }
+
+ if (!anyValid) {
+ LOG.error("no valid quorum servers found in " + HConstants.ZOOKEEPER_CONFIG_NAME);
+ return null;
+ }
+
+ if (clientPort == null) {
+ LOG.error("no clientPort found in " + HConstants.ZOOKEEPER_CONFIG_NAME);
+ return null;
+ }
+
+ if (servers.isEmpty()) {
+ LOG.fatal("No server.X lines found in conf/zoo.cfg. HBase must have a " +
+ "ZooKeeper cluster configured for its operation.");
+ return null;
+ }
+
+ StringBuilder hostPortBuilder = new StringBuilder();
+ for (int i = 0; i < servers.size(); ++i) {
+ String host = servers.get(i);
+ if (i > 0) {
+ hostPortBuilder.append(',');
+ }
+ hostPortBuilder.append(host);
+ hostPortBuilder.append(':');
+ hostPortBuilder.append(clientPort);
+ }
+
+ return hostPortBuilder.toString();
+ }
+
+ /**
* Parse ZooKeeper's zoo.cfg, injecting HBase Configuration variables in.
* This method is used for testing so we can pass our own InputStream.
* @param conf HBaseConfiguration to use for injecting variables.
@@ -232,7 +297,8 @@ public class HQuorumPeer implements HCon
try {
properties.load(inputStream);
} catch (IOException e) {
- String msg = "fail to read properties from " + ZOOKEEPER_CONFIG_NAME;
+ final String msg = "fail to read properties from "
+ + HConstants.ZOOKEEPER_CONFIG_NAME;
LOG.fatal(msg);
throw new IOException(msg, e);
}
@@ -269,8 +335,8 @@ public class HQuorumPeer implements HCon
}
// Special case for 'hbase.cluster.distributed' property being 'true'
if (key.startsWith("server.")) {
- if(conf.get(CLUSTER_DISTRIBUTED).equals(CLUSTER_IS_DISTRIBUTED) &&
- value.startsWith("localhost")) {
+ if (conf.get(HConstants.CLUSTER_DISTRIBUTED).equals(HConstants.CLUSTER_IS_DISTRIBUTED)
+ && value.startsWith("localhost")) {
String msg = "The server in zoo.cfg cannot be set to localhost " +
"in a fully-distributed setup because it won't be reachable. " +
"See \"Getting Started\" for more information.";
Modified: hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKServerTool.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKServerTool.java?rev=1176177&r1=1176176&r2=1176177&view=diff
==============================================================================
--- hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKServerTool.java (original)
+++ hbase/branches/0.89/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKServerTool.java Tue Sep 27 02:41:56 2011
@@ -31,15 +31,16 @@ import java.util.Properties;
* Tool for reading ZooKeeper servers from HBase XML configuation and producing
* a line-by-line list for use by bash scripts.
*/
-public class ZKServerTool implements HConstants {
+public class ZKServerTool {
/**
* Run the tool.
* @param args Command line arguments. First arg is path to zookeepers file.
*/
public static void main(String args[]) {
Configuration conf = HBaseConfiguration.create();
- // Note that we do not simply grab the property ZOOKEEPER_QUORUM from
- // the HBaseConfiguration because the user may be using a zoo.cfg file.
+ // Note that we do not simply grab the property
+ // HConstants.ZOOKEEPER_QUORUM from the HBaseConfiguration because the
+ // user may be using a zoo.cfg file.
Properties zkProps = HQuorumPeer.makeZKProps(conf);
for (Entry<Object, Object> entry : zkProps.entrySet()) {
String key = entry.getKey().toString().trim();