You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ec...@apache.org on 2014/11/25 21:28:57 UTC
[04/19] hbase git commit: HBASE-12476 HydraBase consensus protocol
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InHeapArena.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InHeapArena.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InHeapArena.java
new file mode 100644
index 0000000..617d799
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InHeapArena.java
@@ -0,0 +1,79 @@
+package org.apache.hadoop.hbase.util;
+
+import org.apache.hadoop.hbase.io.hfile.bucket.CacheFullException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A basic Arena Class which handles a chunk of contiguous memory using
+ * a Bucket Allocator.
+ */
+public class InHeapArena implements Arena {
+
+ public static final AtomicInteger numOps = new AtomicInteger();
+ public static final AtomicInteger failedOps = new AtomicInteger();
+ public static final AtomicInteger freedOps = new AtomicInteger();
+
+ public static final Logger LOG = LoggerFactory
+ .getLogger(InHeapArena.class);
+
+ private final byte[] buffer;
+ private BucketAllocator allocator;
+
+ public InHeapArena(int[] bucketSizes, int capacity) {
+ try {
+ allocator = new BucketAllocator(bucketSizes, capacity);
+ } catch (BucketAllocatorException e) {
+ LOG.error("Cannot initialize allocator with size " + capacity +
+ ". Reason", e);
+ }
+ buffer = new byte[capacity];
+ }
+
+ @Override
+ public MemoryBuffer allocateByteBuffer(int size) throws
+ CacheFullException, BucketAllocatorException {
+ int offset;
+
+ try {
+ offset = (int) allocator.allocateBlock(size);
+ numOps.incrementAndGet();
+ } catch (CacheFullException | BucketAllocatorException e) {
+ failedOps.incrementAndGet();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Cannot allocate memory buffer of size " + size, e);
+ }
+ throw e;
+ }
+
+ final ByteBuffer chunk = ByteBuffer.wrap(buffer, offset, size);
+ return new MemoryBuffer(chunk, offset, chunk.remaining());
+ }
+
+ @Override
+ public void freeByteBuffer(final MemoryBuffer buffer) {
+ if (buffer.getOffset() != MemoryBuffer.UNDEFINED_OFFSET) {
+ allocator.freeBlock(buffer.getOffset());
+ freedOps.incrementAndGet();
+ }
+ }
+
+ public static int getNumOpsAndReset() {
+ return numOps.getAndSet(0);
+ }
+
+ public static int getFailedOpsAndReset() {
+ return failedOps.getAndSet(0);
+ }
+
+ public static int getFreedOpsAndReset() {
+ return freedOps.getAndSet(0);
+ }
+
+ public BucketAllocator getAllocator() {
+ return allocator;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InfoServer.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InfoServer.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InfoServer.java
new file mode 100644
index 0000000..4bf49b5
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InfoServer.java
@@ -0,0 +1,148 @@
+/**
+ * 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.util;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.consensus.server.LocalConsensusServer;
+import org.apache.hadoop.http.HttpServer;
+import org.mortbay.jetty.handler.ContextHandlerCollection;
+import org.mortbay.jetty.servlet.Context;
+import org.mortbay.jetty.servlet.DefaultServlet;
+import org.mortbay.jetty.webapp.WebAppContext;
+
+import java.io.IOException;
+import java.net.URL;
+import java.util.Map;
+
+/**
+ * Create a Jetty embedded server to answer http requests. The primary goal
+ * is to serve up status information for the server.
+ * There are three contexts:
+ * "/stacks/" -> points to stack trace
+ * "/static/" -> points to common static files (src/hbase-webapps/static)
+ * "/" -> the jsp server code from (src/hbase-webapps/<name>)
+ */
+public class InfoServer extends HttpServer {
+
+ WebAppContext quorumMonitoring;
+
+ /**
+ * Create a status server on the given port.
+ * The jsp scripts are taken from src/hbase-webapps/<code>name<code>.
+ * @param name The name of the server
+ * @param bindAddress address to bind to
+ * @param port The port to use on the server
+ * @param findPort whether the server should start at the given port and
+ * increment by 1 until it finds a free port.
+ * @throws IOException e
+ */
+ public InfoServer(String name, String bindAddress, int port,
+ boolean findPort, Configuration conf) throws IOException {
+ super(name, bindAddress, port, findPort, conf);
+ webServer.addHandler(new ContextHandlerCollection());
+ }
+
+ protected void addDefaultApps(ContextHandlerCollection parent, String appDir)
+ throws IOException {
+ super.addDefaultApps(parent, appDir, null);
+ // Must be same as up in hadoop.
+ final String logsContextPath = "/logs";
+ // Now, put my logs in place of hadoops... disable old one first.
+ Context oldLogsContext = null;
+ for (Map.Entry<Context, Boolean> e : defaultContexts.entrySet()) {
+ if (e.getKey().getContextPath().equals(logsContextPath)) {
+ oldLogsContext = e.getKey();
+ break;
+ }
+ }
+ if (oldLogsContext != null) {
+ this.defaultContexts.put(oldLogsContext, Boolean.FALSE);
+ }
+ // Now do my logs.
+ // set up the context for "/logs/" if "hadoop.log.dir" property is defined.
+ String logDir = System.getProperty("hbase.log.dir");
+ if (logDir != null) {
+ Context logContext = new Context(parent, "/logs");
+ logContext.setResourceBase(logDir);
+ logContext.addServlet(DefaultServlet.class, "/");
+ defaultContexts.put(logContext, true);
+ }
+ // Now bring up the task monitor
+ WebAppContext taskMonitorContext =
+ new WebAppContext(parent, "taskmontior", "/taskmonitor");
+ taskMonitorContext.addServlet(DefaultServlet.class, "/");
+ taskMonitorContext.setWar(appDir + "/taskmonitor");
+ defaultContexts.put(taskMonitorContext, true);
+
+ // Add some Quorum monitoring page
+ quorumMonitoring = new WebAppContext(parent,
+ HConstants.QUORUM_MONITORING_PAGE_KEY,
+ HConstants.QUORUM_MONITORING_PAGE_PATH);
+ quorumMonitoring.addServlet(DefaultServlet.class, "/");
+ quorumMonitoring.setWar(appDir + HConstants.QUORUM_MONITORING_PAGE_PATH);
+ defaultContexts.put(quorumMonitoring, true);
+ }
+
+ public void addQuorumMonitoringContext(LocalConsensusServer server) {
+ quorumMonitoring.setAttribute(HConstants.QUORUM_MONITORING_PAGE_KEY, server);
+ }
+
+ /**
+ * Get the pathname to the <code>path</code> files.
+ * @return the pathname as a URL
+ */
+ /*
+ protected String getWebAppsPath(String ignore) throws IOException {
+ // Hack: webapps is not a unique enough element to find in CLASSPATH
+ // We'll more than likely find the hadoop webapps dir. So, instead
+ // look for the 'master' webapp in the webapps subdir. That should
+ // get us the hbase context. Presumption is that place where the
+ // master webapp resides is where we want this InfoServer picking up
+ // web applications.
+ final String master = "master";
+ String p = getWebAppDir(master);
+ // Now strip master + the separator off the end of our context
+ return p.substring(0, p.length() - (master.length() + 1/* The separator* /));
+ }
+*/
+
+ private static String getWebAppsPathStatic(final String path)
+ throws IOException {
+ URL url = InfoServer.class.getClassLoader().getResource(path);
+ if (url == null)
+ throw new IOException("hbase-webapps not found in CLASSPATH: " + path);
+ return url.toString();
+ }
+
+ /**
+ * Get the path for this web app
+ * @param webappName web app
+ * @return path
+ * @throws IOException e
+ */
+ public static String getWebAppDir(final String webappName)
+ throws IOException {
+ String webappDir;
+ webappDir = getWebAppsPathStatic("hbase-webapps/" + webappName);
+ return webappDir;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java
new file mode 100644
index 0000000..d550365
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionEvent.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 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.util;
+
+/**
+ * Enumeration of all injection events.
+ * When defining new events, please PREFIX the name
+ * with the supervised class.
+ *
+ * Please see InjectionHandler.
+ */
+public enum InjectionEvent {
+ HMASTER_CREATE_TABLE,
+ HMASTER_DELETE_TABLE,
+ HMASTER_ALTER_TABLE,
+ HMASTER_ENABLE_TABLE,
+ HMASTER_DISABLE_TABLE,
+ ZKUNASSIGNEDWATCHER_REGION_OPENED,
+ SPLITLOGWORKER_SPLIT_LOG_START,
+ HMASTER_START_PROCESS_DEAD_SERVER,
+ HREGIONSERVER_REPORT_RESPONSE,
+
+ // Injection into Store.java
+ READONLYSTORE_COMPACTION_WHILE_SNAPSHOTTING,
+ STORESCANNER_COMPACTION_RACE,
+ STOREFILE_AFTER_WRITE_CLOSE,
+ STOREFILE_AFTER_RENAME
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionHandler.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionHandler.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionHandler.java
new file mode 100644
index 0000000..ff670bc
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/InjectionHandler.java
@@ -0,0 +1,171 @@
+/*
+ * Copyright 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.util;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * The InjectionHandler is an object provided to a class,
+ * which can perform custom actions for JUnit testing.
+ * JUnit test can implement custom version of the handler.
+ * For example, let's say we want to supervise FSImage object:
+ *
+ * <code>
+ * // JUnit test code
+ * class MyInjectionHandler extends InjectionHandler {
+ * protected void _processEvent(InjectionEvent event,
+ * Object... args) {
+ * if (event == InjectionEvent.MY_EVENT) {
+ * LOG.info("Handling my event for fsImage: "
+ * + args[0].toString());
+ * }
+ * }
+ * }
+ *
+ * public void testMyEvent() {
+ * InjectionHandler ih = new MyInjectionHandler();
+ * InjectionHandler.set(ih);
+ * ...
+ *
+ * InjectionHandler.clear();
+ * }
+ *
+ * // supervised code example
+ *
+ * class FSImage {
+ *
+ * private doSomething() {
+ * ...
+ * if (condition1 && InjectionHandler.trueCondition(MY_EVENT1) {
+ * ...
+ * }
+ * if (condition2 || condition3
+ * || InjectionHandler.falseCondition(MY_EVENT1) {
+ * ...
+ * }
+ * ...
+ * InjectionHandler.processEvent(MY_EVENT2, this)
+ * ...
+ * try {
+ * read();
+ * InjectionHandler.processEventIO(MY_EVENT3, this, object);
+ * // might throw an exception when testing
+ * catch (IOEXception) {
+ * LOG.info("Exception")
+ * }
+ * ...
+ * }
+ * ...
+ * }
+ * </code>
+ *
+ * Each unit test should use a unique event type.
+ * The types can be defined by adding them to
+ * InjectionEvent class.
+ *
+ * methods:
+ *
+ * // simulate actions
+ * void processEvent()
+ * // simulate exceptions
+ * void processEventIO() throws IOException
+ *
+ * // simulate conditions
+ * boolean trueCondition()
+ * boolean falseCondition()
+ *
+ * The class implementing InjectionHandler must
+ * override respective protected methods
+ * _processEvent()
+ * _processEventIO()
+ * _trueCondition()
+ * _falseCondition()
+ */
+public class InjectionHandler {
+
+ private static final Log LOG = LogFactory.getLog(InjectionHandler.class);
+
+ // the only handler to which everyone reports
+ private static InjectionHandler handler = new InjectionHandler();
+
+ // can not be instantiated outside, unless a testcase extends it
+ protected InjectionHandler() {}
+
+ // METHODS FOR PRODUCTION CODE
+
+ protected void _processEvent(InjectionEvent event, Object... args) {
+ // by default do nothing
+ }
+
+ protected void _processEventIO(InjectionEvent event, Object... args) throws IOException{
+ // by default do nothing
+ }
+
+ protected boolean _trueCondition(InjectionEvent event, Object... args) {
+ return true; // neutral in conjunction
+ }
+
+ protected boolean _falseCondition(InjectionEvent event, Object... args) {
+ return false; // neutral in alternative
+ }
+
+ ////////////////////////////////////////////////////////////
+
+ /**
+ * Set to the empty/production implementation.
+ */
+ public static void clear() {
+ handler = new InjectionHandler();
+ }
+
+ /**
+ * Set custom implementation of the handler.
+ */
+ public static void set(InjectionHandler custom) {
+ LOG.warn("WARNING: SETTING INJECTION HANDLER" +
+ " - THIS SHOULD NOT BE USED IN PRODUCTION !!!");
+ handler = custom;
+ }
+
+ /*
+ * Static methods for reporting to the handler
+ */
+
+ public static void processEvent(InjectionEvent event, Object... args) {
+ handler._processEvent(event, args);
+ }
+
+ public static void processEventIO(InjectionEvent event, Object... args)
+ throws IOException {
+ handler._processEventIO(event, args);
+ }
+
+ public static boolean trueCondition(InjectionEvent event, Object... args) {
+ return handler._trueCondition(event, args);
+ }
+
+ public static boolean falseCondition(InjectionEvent event, Object... args) {
+ return handler._falseCondition(event, args);
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/MemoryBuffer.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/MemoryBuffer.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/MemoryBuffer.java
new file mode 100644
index 0000000..7dfc814
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/MemoryBuffer.java
@@ -0,0 +1,48 @@
+package org.apache.hadoop.hbase.util;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Wraps the memory allocated by the Arena. It provides the abstraction of a
+ * ByteBuffer to the outside world.
+ */
+public class MemoryBuffer {
+
+ private final long offset;
+ private final int size;
+ private final ByteBuffer buffer;
+ public static final int UNDEFINED_OFFSET = -1;
+
+ public MemoryBuffer(final ByteBuffer buffer, long offset, int size) {
+ this.offset = offset;
+ this.size = size;
+ this.buffer = buffer;
+ }
+
+ public MemoryBuffer(ByteBuffer buffer) {
+ this.offset = UNDEFINED_OFFSET;
+ this.size = buffer.remaining();
+ this.buffer = buffer;
+ }
+
+ public ByteBuffer getBuffer() {
+ return buffer;
+ }
+
+ public long getOffset() {
+ return offset;
+ }
+
+ public int getSize() {
+ return size;
+ }
+
+ public void flip() {
+ if (offset != UNDEFINED_OFFSET) {
+ buffer.limit(buffer.position());
+ buffer.position((int)offset);
+ } else {
+ buffer.flip();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Pair.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Pair.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Pair.java
new file mode 100644
index 0000000..8e9fa36
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Pair.java
@@ -0,0 +1,131 @@
+/**
+ * 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.util;
+
+import java.io.Serializable;
+
+/**
+ * A generic class for pairs.
+ * @param <T1>
+ * @param <T2>
+ */
+public class Pair<T1, T2> implements Serializable
+{
+ private static final long serialVersionUID = -3986244606585552569L;
+ protected T1 first = null;
+ protected T2 second = null;
+
+ /**
+ * Default constructor.
+ */
+ public Pair()
+ {
+ }
+
+ /**
+ * Constructor
+ * @param a operand
+ * @param b operand
+ */
+ public Pair(T1 a, T2 b)
+ {
+ this.first = a;
+ this.second = b;
+ }
+
+ /**
+ * Constructs a new pair, inferring the type via the passed arguments
+ * @param <T1> type for first
+ * @param <T2> type for second
+ * @param a first element
+ * @param b second element
+ * @return a new pair containing the passed arguments
+ */
+ public static <T1,T2> Pair<T1,T2> newPair(T1 a, T2 b) {
+ return new Pair<T1,T2>(a, b);
+ }
+
+ /**
+ * Replace the first element of the pair.
+ * @param a operand
+ */
+ public void setFirst(T1 a)
+ {
+ this.first = a;
+ }
+
+ /**
+ * Replace the second element of the pair.
+ * @param b operand
+ */
+ public void setSecond(T2 b)
+ {
+ this.second = b;
+ }
+
+ /**
+ * Return the first element stored in the pair.
+ * @return T1
+ */
+ public T1 getFirst()
+ {
+ return first;
+ }
+
+ /**
+ * Return the second element stored in the pair.
+ * @return T2
+ */
+ public T2 getSecond()
+ {
+ return second;
+ }
+
+ private static boolean equals(Object x, Object y)
+ {
+ return (x == null && y == null) || (x != null && x.equals(y));
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean equals(Object other)
+ {
+ return other instanceof Pair && equals(first, ((Pair)other).first) &&
+ equals(second, ((Pair)other).second);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ if (first == null)
+ return (second == null) ? 0 : second.hashCode() + 1;
+ else if (second == null)
+ return first.hashCode() + 2;
+ else
+ return first.hashCode() * 17 + second.hashCode();
+ }
+
+ @Override
+ public String toString()
+ {
+ return "{" + getFirst() + "," + getSecond() + "}";
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Threads.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Threads.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Threads.java
new file mode 100644
index 0000000..4094bc7
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/Threads.java
@@ -0,0 +1,264 @@
+/**
+ * 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.util;
+
+import java.io.InterruptedIOException;
+import java.io.PrintWriter;
+import java.lang.Thread.UncaughtExceptionHandler;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.hbase.util.HasThread;
+
+/**
+ * Thread Utility
+ */
+public class Threads {
+ protected static final Log LOG = LogFactory.getLog(Threads.class);
+
+ /**
+ * Utility method that sets name, daemon status and starts passed thread.
+ * @param t thread to frob
+ * @param name new name
+ * @return Returns the passed Thread <code>t</code>.
+ */
+ public static Thread setDaemonThreadRunning(final Thread t,
+ final String name) {
+ return setDaemonThreadRunning(t, name, null);
+ }
+
+ public static HasThread setDaemonThreadRunning(final HasThread t,
+ final String name) {
+ setDaemonThreadRunning(t.getThread(), name, null);
+ return t;
+ }
+
+ /**
+ * Utility method that sets name, daemon status and starts passed thread.
+ * @param t thread to frob
+ * @param name new name
+ * @param handler A handler to set on the thread. Pass null if want to
+ * use default handler.
+ * @return Returns the passed Thread <code>t</code>.
+ */
+ public static Thread setDaemonThreadRunning(final Thread t,
+ final String name, final UncaughtExceptionHandler handler) {
+ if (name != null) {
+ t.setName(name);
+ }
+ if (handler != null) {
+ t.setUncaughtExceptionHandler(handler);
+ }
+ t.setDaemon(true);
+ t.start();
+ return t;
+ }
+
+ public static HasThread setDaemonThreadRunning(final HasThread t,
+ final String name, final UncaughtExceptionHandler handler) {
+ setDaemonThreadRunning(t.getThread(), name, handler);
+ return t;
+ }
+
+ /**
+ * Shutdown passed thread using isAlive and join.
+ * @param t Thread to shutdown
+ */
+ public static void shutdown(final Thread t) {
+ shutdown(t, 0);
+ }
+
+ public static void shutdown(final HasThread t) {
+ shutdown(t.getThread(), 0);
+ }
+
+ /**
+ * Shutdown passed thread using isAlive and join.
+ * @param joinwait Pass 0 if we're to wait forever.
+ * @param t Thread to shutdown
+ */
+ public static void shutdown(final Thread t, final long joinwait) {
+ if (t == null) return;
+ while (t.isAlive()) {
+ try {
+ t.join(joinwait);
+ } catch (InterruptedException e) {
+ LOG.warn(t.getName() + "; joinwait=" + joinwait, e);
+ }
+ }
+ }
+
+ public static void shutdown(final HasThread t, final long joinwait) {
+ shutdown(t.getThread(), joinwait);
+ }
+
+ /**
+ * @param t Waits on the passed thread to die dumping a threaddump every
+ * minute while its up.
+ * @throws InterruptedException
+ */
+ public static void threadDumpingIsAlive(final Thread t)
+ throws InterruptedException {
+ if (t == null) {
+ return;
+ }
+ long startTime = System.currentTimeMillis();
+ while (t.isAlive()) {
+ Thread.sleep(1000);
+ if (System.currentTimeMillis() - startTime > 60000) {
+ startTime = System.currentTimeMillis();
+ ReflectionUtils.printThreadInfo(new PrintWriter(System.out),
+ "Automatic Stack Trace every 60 seconds waiting on " +
+ t.getName());
+ }
+ }
+ }
+
+ /**
+ * @param millis How long to sleep for in milliseconds.
+ */
+ public static void sleep(long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted when sleeping for " + millis + "ms, ignoring", e);
+ }
+ }
+
+ /**
+ * Sleeps for the given amount of time. Retains the thread's interruption status.
+ * @param millis How long to sleep for in milliseconds.
+ */
+ public static void sleepRetainInterrupt(long millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted when sleeping for " + millis + "ms");
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Sleeps for the given amount of time even if interrupted. Preserves
+ * the interrupt status.
+ * @param msToWait the amount of time to sleep in milliseconds
+ */
+ public static void sleepWithoutInterrupt(final long msToWait) {
+ long timeMillis = System.currentTimeMillis();
+ long endTime = timeMillis + msToWait;
+ boolean interrupted = false;
+ while (timeMillis < endTime) {
+ try {
+ Thread.sleep(endTime - timeMillis);
+ } catch (InterruptedException ex) {
+ interrupted = true;
+ }
+ timeMillis = System.currentTimeMillis();
+ }
+
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Create a new CachedThreadPool with a bounded number as the maximum
+ * thread size in the pool.
+ *
+ * @param maxCachedThread the maximum thread could be created in the pool
+ * @param timeout the maximum time to wait
+ * @param unit the time unit of the timeout argument
+ * @param threadFactory the factory to use when creating new threads
+ * @return threadPoolExecutor the cachedThreadPool with a bounded number
+ * as the maximum thread size in the pool.
+ */
+ public static ThreadPoolExecutor getBoundedCachedThreadPool(
+ int maxCachedThread, long timeout, TimeUnit unit,
+ ThreadFactory threadFactory) {
+ ThreadPoolExecutor boundedCachedThreadPool =
+ new ThreadPoolExecutor(maxCachedThread, maxCachedThread, timeout,
+ TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory);
+ // allow the core pool threads timeout and terminate
+ boundedCachedThreadPool.allowCoreThreadTimeOut(true);
+ return boundedCachedThreadPool;
+ }
+
+ /**
+ * Creates a ThreadPoolExecutor which has a bound on the number of tasks that can be
+ * submitted to it, determined by the blockingLimit parameter. Excess tasks
+ * submitted will block on the calling thread till space frees up.
+ *
+ * @param blockingLimit max number of tasks that can be submitted
+ * @param timeout time value after which unused threads are killed
+ * @param unit time unit for killing unused threads
+ * @param threadFactory thread factory to use to spawn threads
+ * @return the ThreadPoolExecutor
+ */
+ public static ThreadPoolExecutor getBlockingThreadPool(
+ int blockingLimit, long timeout, TimeUnit unit,
+ ThreadFactory threadFactory) {
+ ThreadPoolExecutor blockingThreadPool =
+ new ThreadPoolExecutor(
+ 1, blockingLimit, timeout, TimeUnit.SECONDS,
+ new SynchronousQueue<Runnable>(),
+ threadFactory,
+ new RejectedExecutionHandler() {
+ @Override
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+ try {
+ // The submitting thread will block until the thread pool frees up.
+ executor.getQueue().put(r);
+ } catch (InterruptedException e) {
+ throw new RejectedExecutionException(
+ "Failed to requeue the rejected request because of ", e);
+ }
+ }
+ });
+ blockingThreadPool.allowCoreThreadTimeOut(true);
+ return blockingThreadPool;
+ }
+
+ public static void renameThread(Thread t, String newName) {
+ String oldName = t.getName();
+ if (!oldName.equals(newName)) {
+ LOG.info("Thread '" + oldName + "' is now known as '" + newName + "'");
+ t.setName(newName);
+ }
+ }
+
+ /**
+ * A helper function to check if the current thread was interrupted.
+ *
+ * @throws InterruptedIOException if the thread had been interrupted.
+ **/
+ public static void checkInterrupted() throws InterruptedIOException {
+ if (Thread.interrupted()) {
+ throw new InterruptedIOException("Thread was interrupted");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/AsyncSerialExecutorServiceImpl.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/AsyncSerialExecutorServiceImpl.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/AsyncSerialExecutorServiceImpl.java
new file mode 100644
index 0000000..9ea0f4b
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/AsyncSerialExecutorServiceImpl.java
@@ -0,0 +1,101 @@
+package org.apache.hadoop.hbase.util.serial;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.util.DaemonThreadFactory;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+
+/**
+ * Library that can be used when we want to schedule many async events but such that execution of a
+ * certain event depends on completing a previous event. One doesn't need to know on which thread
+ * the execution will happen - since the next free thread from the pool will be chosen and the
+ * constraint will be preserved.
+ */
+public class AsyncSerialExecutorServiceImpl implements SerialExecutorService {
+ static final Log LOG = LogFactory.getLog(AsyncSerialExecutorServiceImpl.class);
+
+ private final Executor executorService;
+
+ public AsyncSerialExecutorServiceImpl(int numThreads, String poolName) {
+ this.executorService =
+ Executors.newFixedThreadPool(numThreads, new DaemonThreadFactory(poolName));
+ }
+
+ public AsyncSerialExecutorServiceImpl(Executor executorService) {
+ this.executorService = executorService;
+ }
+
+ @Override
+ public SerialExecutionStream createStream() {
+ return new SerialExecutionStream() {
+ /**
+ * Future that completes when last scheduled action finishes, representing last
+ * future in the stream/list.
+ * So, each future needs to have only one listener attached to it.
+ * We enforce that by accessing lastExecuted only through getAndSet method,
+ * which guarantees each future is both stored and returned by this AtomicReference
+ * only once.
+ *
+ * At start it has already been finished, so first command doesn't have to wait.
+ */
+ private AtomicReference<ListenableFuture<Void>> lastExecuted = new AtomicReference<>(
+ Futures.<Void>immediateFuture(null));
+
+ @Override
+ public ListenableFuture<Void> execute(Callable<ListenableFuture<Void>> command) {
+ SettableFuture<Void> commandDone = SettableFuture.create();
+ ListenableFuture<Void> last = lastExecuted.getAndSet(commandDone);
+ last.addListener(new CommandRunAndCompleteCurrent(command, commandDone), executorService);
+ return commandDone;
+ }
+ };
+ }
+
+ /**
+ * Wrapper around runnable and settable future
+ */
+ private class CommandRunAndCompleteCurrent implements Runnable {
+ private Callable<ListenableFuture<Void>> command;
+ private SettableFuture<Void> currentFuture;
+
+ public CommandRunAndCompleteCurrent(Callable<ListenableFuture<Void>> command, SettableFuture<Void> currentFuture) {
+ this.command = command;
+ this.currentFuture = currentFuture;
+ }
+
+ @Override
+ public void run() {
+ try {
+ ListenableFuture<Void> commandFuture = command.call();
+ if (commandFuture == null) {
+ currentFuture.set(null);
+ } else {
+ Futures.addCallback(commandFuture, new FutureCallback<Void>() {
+ @Override
+ public void onSuccess(Void result) {
+ currentFuture.set(null);
+ }
+ @Override
+ public void onFailure(Throwable t) {
+ LOG.error("exception in the command future", t);
+ currentFuture.set(null);
+ }
+ }, executorService);
+ }
+ } catch (Exception e) {
+ // Log exception
+ LOG.error("exception while executing command", e);
+ currentFuture.set(null);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/SerialExecutorService.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/SerialExecutorService.java b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/SerialExecutorService.java
new file mode 100644
index 0000000..3f26b36
--- /dev/null
+++ b/hbase-consensus/src/main/java/org/apache/hadoop/hbase/util/serial/SerialExecutorService.java
@@ -0,0 +1,34 @@
+package org.apache.hadoop.hbase.util.serial;
+
+import java.util.concurrent.Callable;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+/**
+ * Executor service which allows creation of arbitrary number of serial streams, all to be executed
+ * on a single thread pool. All commands scheduled on a single serial stream are executed serially
+ * (not necessarily on the same thread).
+ */
+public interface SerialExecutorService {
+ /** Creates new serial stream */
+ SerialExecutionStream createStream();
+
+ /**
+ * Executor, that makes sure commands are executed serially, in
+ * order they are scheduled. No two commands will be running at the same time.
+ */
+ public interface SerialExecutionStream {
+ /**
+ * Executes the given command on this stream, returning future.
+ * If future command return is not null, next command in the stream
+ * will not be executed before future completes.
+ *
+ * I.e. async commands should return future when it completes,
+ * and non-async commands should return null.
+ *
+ * @return Returns future which is going to be completed when command and
+ * future it returns both finish.
+ */
+ ListenableFuture<Void> execute(Callable<ListenableFuture<Void>> command);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/LocalTestBed.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/LocalTestBed.java b/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/LocalTestBed.java
new file mode 100644
index 0000000..9f7a7c3
--- /dev/null
+++ b/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/LocalTestBed.java
@@ -0,0 +1,944 @@
+package org.apache.hadoop.hbase.consensus;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HServerAddress;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.consensus.exceptions.LeaderNotReadyException;
+import org.apache.hadoop.hbase.consensus.exceptions.NewLeaderException;
+import org.apache.hadoop.hbase.consensus.quorum.QuorumAgent;
+import org.apache.hadoop.hbase.consensus.quorum.RaftQuorumContext;
+import org.apache.hadoop.hbase.consensus.server.InstrumentedConsensusServiceImpl;
+import org.apache.hadoop.hbase.consensus.server.LocalConsensusServer;
+import org.apache.hadoop.hbase.consensus.server.peer.states.PeerHandleAppendResponse;
+
+import org.apache.hadoop.hbase.consensus.util.RaftUtil;
+import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A test utility for Unit Testing RAFT protocol by itself.
+ */
+public class LocalTestBed {
+ private static final Logger LOG = LoggerFactory.getLogger(LocalTestBed.class);
+
+ private static int DEFAULT_RAFT_CONSENSUS_START_PORT_NUMBER = 60080;
+ private static int QUORUM_SIZE = 5;
+ private static final int QUORUM_MAJORITY = 3;
+ private static int nextPortNumber;
+
+ private Configuration conf;
+ private static HRegionInfo regionInfo;
+ private static RaftTestUtil RAFT_TEST_UTIL = new RaftTestUtil();
+ private final List<int[]> mockLogs;
+ private boolean checkLeaderCount = false;
+ private boolean usePersistentLog = true;
+ private boolean autoRestartThriftServer = true;
+ private AtomicLong commitSuccessCount = new AtomicLong(0);
+ private AtomicLong commitFailureCount = new AtomicLong(0);
+ private HealthChecker checker = null;
+
+ public static String LOCAL_HOST = "127.0.0.1";
+ ConcurrentHashMap<String, LocalConsensusServer> servers;
+
+ public final long BigBang = System.currentTimeMillis();
+
+ public static class HealthChecker extends Thread {
+ private HRegionInfo regionInfo;
+ private RaftTestUtil testUtil;
+ private boolean autoRestartThriftServer;
+ private long checkInterval;
+ private AtomicBoolean time2die = new AtomicBoolean(false);
+
+ public HealthChecker(HRegionInfo regionInfo, RaftTestUtil testUtil, boolean autoRestartThriftServer, long checkInterval) {
+ this.regionInfo = regionInfo;
+ this.testUtil = testUtil;
+ this.autoRestartThriftServer = autoRestartThriftServer;
+ this.checkInterval = checkInterval;
+ }
+
+ @Override
+ public void run() {
+ long previousCheckTime = 0L;
+ while (!time2die.get()) {
+ try {
+ long now = System.currentTimeMillis();
+ if (now >= previousCheckTime + checkInterval) {
+ LOG.info("checking the health of all quorum members ......");
+ testUtil.checkHealth(regionInfo, autoRestartThriftServer);
+ previousCheckTime = now = System.currentTimeMillis();
+ }
+ long sleepTime = previousCheckTime + checkInterval - now;
+ if (sleepTime > 0) {
+ Thread.sleep(sleepTime);
+ }
+ } catch (InterruptedException ex) {
+ LOG.info("Time to Die!!");
+ time2die.set(true);
+ } catch (Throwable stone) {
+ LOG.warn("Caught an exception while checking health", stone);
+ }
+ }
+ }
+
+ public void shutdown() {
+ time2die.set(true);
+ }
+ }
+
+
+ public static class LinkState {
+ private HServerAddress src;
+ private HServerAddress dst;
+ private State state;
+ private long delay = 0L;
+ private long hiccupStopTime = 0L;
+ private long hiccupStartTime = 0L;
+
+ public static enum State {
+ UP, DOWN
+ }
+
+ public LinkState(HServerAddress src, HServerAddress dst, State state) {
+ this.src = src;
+ this.dst = dst;
+ this.state = state;
+ }
+
+ public HServerAddress getSrc() {
+ return src;
+ }
+
+ public HServerAddress getDst() {
+ return dst;
+ }
+
+ public void setDelay(long delay) {
+ this.delay = delay;
+ }
+
+ public boolean inHiccup(long now) {
+ return hiccupStartTime <= now && now <= hiccupStopTime;
+ }
+
+ public long getDelay(long now) {
+ return delay;
+ }
+
+ public void setHiccupTime(long start, long stop) {
+ hiccupStartTime = start;
+ hiccupStopTime = stop;
+ }
+
+ public long getHiccupStartTime() {
+ return hiccupStartTime;
+ }
+
+ public long getHiccupStopTime() {
+ return hiccupStopTime;
+ }
+
+ @Override
+ public String toString() {
+ return "[" + src + "--" + dst + "]";
+ }
+ }
+
+ public static class NodeState {
+ private State state;
+ public static enum State {
+ UP, DOWN, INTERMITTENT, SLOW
+ }
+
+ public NodeState(State state) {
+ }
+
+ public NodeState changeState(State state) {
+ return this;
+ }
+ }
+
+ public static class Chaos {
+ private TestConfig config;
+ private Random prng;
+
+ // the map is maintained per pair of nodes
+ private ConcurrentHashMap<HServerAddress, ConcurrentHashMap<HServerAddress, LinkState>> links
+ = new ConcurrentHashMap<HServerAddress, ConcurrentHashMap<HServerAddress, LinkState>>();
+
+ private ConcurrentHashMap<String, NodeState> nodes =
+ new ConcurrentHashMap<String, NodeState>();
+
+ public Chaos() {
+ this(new TestConfig());
+ }
+
+ public Chaos(TestConfig config) {
+ this.config = config;
+ prng = new Random(config.getLong("seed", System.currentTimeMillis()));
+ }
+
+ public LinkState getLinkState(HServerAddress src, HServerAddress dst) {
+ if (src == null || dst == null || src.equals(dst)) {
+ return null;
+ }
+ if (src.toString().compareTo(dst.toString()) > 0) {
+ HServerAddress tmp = src;
+ src = dst;
+ dst = tmp;
+ }
+ if (links.get(src) == null) {
+ links.put(src, new ConcurrentHashMap<HServerAddress, LinkState>());
+ }
+ if (links.get(src).get(dst) == null) {
+ links.get(src).put(dst, new LinkState(src, dst, LinkState.State.UP));
+ }
+ return links.get(src).get(dst);
+ }
+
+ public void updateNetworkStates(long now, boolean clear) {
+ double defaultDropRate = config.getDouble("packet-drop-rate", 0.0);
+ double maxRandomDelay = config.getDouble("max-random-delay", 20.0);
+ double minRandomDelay = config.getDouble("min-random-delay", 00.0);
+
+ long maxHiccupDuration = config.getLong("max-hiccup-duration", 10000L);
+ long minHiccupDuration = config.getLong("min-hiccup-duration", 1000L);
+ long maxHiccupLinks = config.getLong("max-hiccup-links", 1L);
+ long hiccupLambda = config.getLong("hiccup-lambda", -1L);
+
+ ConcurrentHashMap<String, LocalConsensusServer> servers = RAFT_TEST_UTIL.getServers();
+ Set<LinkState> currentHiccups = new HashSet<LinkState>();
+ Set<LinkState> allHiccups = new HashSet<LinkState>(); // include future hiccups
+ Set<LinkState> nohiccups = new HashSet<LinkState>();
+ for (HServerAddress dst : regionInfo.getPeersWithRank().keySet()) {
+ dst = RaftUtil.getLocalConsensusAddress(dst);
+ for (HServerAddress src : regionInfo.getPeersWithRank().keySet()) {
+ src = RaftUtil.getLocalConsensusAddress(src);
+ if (src.equals(dst)) {
+ continue;
+ }
+ LinkState link = getLinkState(src, dst);
+ if (link.inHiccup(now)) {
+ currentHiccups.add(link);
+ allHiccups.add(link);
+ } else if (link.getHiccupStartTime() > now) {
+ allHiccups.add(link);
+ } else {
+ // We know link.getHiccupStopTime() < now, too
+ nohiccups.add(link);
+ }
+ }
+ }
+
+ if (currentHiccups.size() > 0) {
+ LOG.debug("The following links are in hiccup: " + currentHiccups);
+ }
+
+ List<LinkState> futureHiccups = new ArrayList<LinkState>(nohiccups);
+
+ // Spread the love; otherwise, the first few links are always in hiccups (now
+ // and in future).
+ Collections.shuffle(futureHiccups);
+
+ for (LinkState link : futureHiccups) {
+ // Calculate the next hiccup time
+ long delta = poissonRandomInterarrivalDelay(prng, (double)hiccupLambda);
+ long hiccupStartTime = link.getHiccupStopTime() == 0 ? now + delta : link.getHiccupStopTime() + delta;
+ long duration = (long)(prng.nextDouble()*(maxHiccupDuration-minHiccupDuration) + minHiccupDuration);
+ long hiccupStopTime = hiccupStartTime + duration;
+
+ // before we schedule this hiccup, make sure we don't have too many
+ // hiccups at the same time (both now and in future).
+ LinkState tmp = new LinkState(link.getSrc(), link.getDst(), LinkState.State.UP);
+ tmp.setHiccupTime(hiccupStartTime, hiccupStopTime);
+ allHiccups.add(tmp);
+ int nhiccups =computeMaxHiccupLinks(allHiccups);
+ allHiccups.remove(tmp);
+
+ if (nhiccups < maxHiccupLinks) {
+ LOG.debug("---- scheduling a future hiccup for " + link + " [" + hiccupStartTime + ", " + hiccupStopTime + "]");
+ link.setHiccupTime(hiccupStartTime, hiccupStopTime);
+ allHiccups.add(link);
+ } else {
+ LOG.debug("---- too many hiccups right now; not scheduling a future hiccup for " + link + " [" + hiccupStartTime + ", " + hiccupStopTime + "]");
+ }
+ }
+
+ for (HServerAddress dst : regionInfo.getPeersWithRank().keySet()) {
+ dst = RaftUtil.getLocalConsensusAddress(dst);
+ InstrumentedConsensusServiceImpl service = (InstrumentedConsensusServiceImpl)
+ (servers.get(dst.getHostAddressWithPort()).getHandler());
+ for (HServerAddress src : regionInfo.getPeersWithRank().keySet()) {
+ src = RaftUtil.getLocalConsensusAddress(src);
+ if (!src.equals(dst) && !clear) {
+ long delay = (long)(prng.nextDouble()*(maxRandomDelay-minRandomDelay) + minRandomDelay);
+ double dropRate = defaultDropRate;
+ boolean inHiccup = false;
+ if (hiccupLambda > 0) {
+ LinkState link = getLinkState(src, dst);
+ if (link.inHiccup(now)) {
+ LOG.debug("---- " + link + " in hiccup right now!");
+ dropRate = 1.1; // make sure we take care of floating-point inaccuracies
+ inHiccup = true;
+ }
+ }
+ service.setHiccup(src.toString(), inHiccup);
+ service.setPacketDropRate(src.toString(), dropRate);
+ service.setPacketDelay(src.toString(), delay);
+ } else {
+ service.setHiccup(src.toString(), false);
+ service.setPacketDropRate(src.toString(), 0.0);
+ service.setPacketDelay(src.toString(), 0L);
+ }
+ }
+ }
+ }
+
+ /**
+ * computeMaxHiccupLinks
+ *
+ * Compute the maximum number of overlapping hiccups.
+ *
+ * We can really use a unit test here.
+ */
+ public int computeMaxHiccupLinks(Set<LinkState> hiccups) {
+ if (hiccups == null || hiccups.size() == 0) {
+ return 0;
+ }
+ List<Long> endpoints = new ArrayList<Long>();
+ for (LinkState x : hiccups) {
+ endpoints.add(x.getHiccupStartTime());
+ endpoints.add(x.getHiccupStopTime());
+ }
+ Collections.sort(endpoints);
+
+ // we are guaranteed here that endpoints.size() >= 2
+ int maxCount = 0;
+ for (int i=1; i<endpoints.size(); i++) {
+ Long start = endpoints.get(i-1);
+ Long stop = endpoints.get(i);
+ int count = 0;
+ for (LinkState link : hiccups) {
+ // Two intervals are disjoint iff one's start > the other's end
+ // because there exists a point that partitions the one-dimensional
+ // space.
+ boolean disjoint = link.getHiccupStartTime() > stop
+ || start > link.getHiccupStopTime();
+ if (!disjoint) {
+ count ++;
+ }
+ }
+ maxCount = Math.max(maxCount, count);
+ }
+ return maxCount;
+ }
+ }
+
+ public static class Adversary {
+ private Chaos chaos;
+ private long previousNetworkChangeTime = 0L;
+ private TestConfig config;
+
+ public Adversary(Chaos chaos, TestConfig config) {
+ this.chaos = chaos;
+ this.config = config;
+ }
+
+ public Chaos getChaos() {
+ return chaos;
+ }
+
+ public List<TestEvent> getEvents(long now) {
+ if (now - previousNetworkChangeTime > 1000L) {
+ LOG.info("Changing network delays ......");
+ chaos.updateNetworkStates(now, false);
+ previousNetworkChangeTime = now;
+ }
+ return new ArrayList<TestEvent>();
+ }
+ }
+
+
+ public static class TestEvent {
+ private EventType type;
+ private long when;
+ private List<String> nodes;
+ private List<WALEdit> edits;
+ private long sleepAmount;
+ private Map<String, Map<String, Long>> delayMap;
+
+ public static enum EventType {
+ START, STOP, NOOP, SLEEP, NEW_COMMITS, UPDATE_DELAYS, CRASH_NODES, START_NODES, DONE;
+ }
+
+ public TestEvent(EventType type) {
+ this(type, System.currentTimeMillis());
+ }
+
+ public TestEvent(EventType type, long when) {
+ this.type = type;
+ this.when = when;
+ }
+
+ public EventType type() {
+ return type;
+ }
+
+ public long when() {
+ return when;
+ }
+
+ public long sleepAmount() {
+ return sleepAmount;
+ }
+
+ public List<String> nodes() {
+ return nodes;
+ }
+
+ public Map<String, Map<String, Long>> delayMap() {
+ return delayMap;
+ }
+
+ public List<WALEdit> edits() {
+ return edits;
+ }
+
+ public TestEvent sleep(long ms) {
+ this.sleepAmount = ms;
+ return this;
+ }
+
+ public TestEvent crash(List<String> nodes) {
+ this.nodes = nodes;
+ return this;
+ }
+
+ public TestEvent start(List<String> nodes) {
+ this.nodes = nodes;
+ return this;
+ }
+
+ public TestEvent updateDelays(Map<String, Map<String, Long>> delayMap) {
+ this.delayMap = delayMap;
+ return this;
+ }
+
+ public TestEvent newCommits(List<WALEdit> edits) {
+ this.edits = edits;
+ return this;
+ }
+ }
+
+ public static class TestRequest implements Runnable {
+ public static enum State {
+ READY, RUNNING, DONE, FAILED
+ }
+ private LocalTestBed testbed;
+ private List<WALEdit> edits;
+ private State state = State.READY;
+
+ public TestRequest(LocalTestBed testbed, List<WALEdit> edits) {
+ this.testbed = testbed;
+ this.edits = edits;
+ }
+
+ public synchronized State state() {
+ return state;
+ }
+
+ public void run() {
+ synchronized (this) { state = State.RUNNING; }
+ LOG.info("commiting ......");
+ testbed.dumpStates();
+ boolean done = testbed.testSingleCommit(edits);
+ synchronized (this) { state = done ? State.DONE : State.FAILED; }
+ testbed.incrementCommitCount(done);
+ LOG.info("DONE commiting ......");
+ testbed.dumpStates();
+ }
+ }
+
+
+ public static class TestConfig {
+
+ private Map<String, String> kvs = new HashMap<String, String>();
+
+ public TestConfig(String... args) {
+ kvs.put("seed", "" + System.currentTimeMillis());
+ kvs.put("packet-arrival-lambda", "100.0");
+ kvs.put("max-test-time", "3600000");
+ kvs.put("max-commits", "100");
+ kvs.put("event-loop-step", "1");
+ kvs.put("min-random-delay", "10");
+ kvs.put("max-random-delay", "100");
+ kvs.put("packet-drop-rate", "0.00");
+
+ for (String arg : args) {
+ String [] pair = arg.trim().split("=");
+ if (pair.length == 2) {
+ kvs.put(pair[0], pair[1]);
+ } else if (pair.length == 1) {
+ kvs.put(pair[0], "");
+ }
+ }
+ }
+
+ public String get(String key) {
+ return kvs.get(key);
+ }
+
+ public String get(String key, String defaultValue) {
+ if (kvs.get(key) == null) {
+ return defaultValue;
+ } else {
+ return kvs.get(key);
+ }
+ }
+
+ public double getDouble(String key, double defaultValue) {
+ String value = kvs.get(key);
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return Double.parseDouble(value);
+ } catch (Exception ex) {
+ return defaultValue;
+ }
+ }
+
+ public long getLong(String key, long defaultValue) {
+ String value = kvs.get(key);
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return Long.parseLong(value);
+ } catch (Exception ex) {
+ return defaultValue;
+ }
+ }
+
+ public int getInt(String key, int defaultValue) {
+ String value = kvs.get(key);
+ if (value == null) {
+ return defaultValue;
+ }
+ try {
+ return Integer.parseInt(value);
+ } catch (Exception ex) {
+ return defaultValue;
+ }
+ }
+ }
+
+ public static class Test implements Runnable {
+ private LocalTestBed testbed;
+ private TestConfig config;
+ private long testStartTime;
+ private long previousTick;
+ private int ncommits = 0;
+ private Adversary adversary;
+ private long testEndTime;
+ private long nextCommitTime;
+ private Random prng;
+
+ public Test(TestConfig config, LocalTestBed testbed) {
+ this.config = config;
+ this.testbed = testbed;
+ }
+
+ public void init(Adversary adversary) {
+ prng = new Random(config.getLong("seed", System.currentTimeMillis()));
+ testStartTime = System.currentTimeMillis();
+ previousTick = 0L;
+ this.adversary = adversary;
+ nextCommitTime = testStartTime + 5000L; // one second to settle
+ LOG.info("nextCommitTime = " + nextCommitTime);
+ }
+
+ public void incrementSuccessfulCommits() {
+ }
+
+ public List<TestEvent> getNextEvents(long now) {
+ List<TestEvent> events = new ArrayList<TestEvent>();
+ long wakeupTime = previousTick + config.getLong("event-loop-step", 50L);
+ long delay = Math.max(wakeupTime, now) - now;
+ if (delay > 0) {
+ noThrowSleep(delay);
+ }
+ previousTick = Math.max(wakeupTime, now);
+
+ for (TestEvent event : adversary.getEvents(now)) {
+ events.add(event);
+ }
+ // flip a coin
+ if (now >= nextCommitTime) {
+ events.add(new TestEvent(TestEvent.EventType.NEW_COMMITS).newCommits(generateTestingWALEdit()));
+ nextCommitTime += poissonRandomInterarrivalDelay(prng, config.getDouble("packet-arrival-lambda", 100));
+ ncommits ++;
+ }
+ return events;
+ }
+
+ public void run() {
+ BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>();
+ ThreadPoolExecutor fakeClient = new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, workQueue);
+ long previousStateDumpTime = 0L;
+ List<TestRequest> requests = new ArrayList<TestRequest>();
+ List<TestRequest> processedRequests = new ArrayList<TestRequest>();
+ mainloop:
+ while ((long)processedRequests.size() < config.getLong("max-commits", 10L)) {
+ List<TestEvent> events = getNextEvents(System.currentTimeMillis());
+ for (TestEvent event : events) {
+ switch (event.type()) {
+ case SLEEP:
+ noThrowSleep(event.sleepAmount());
+ break;
+
+ case NEW_COMMITS:
+ if (requests.size() <= 0) {
+ TestRequest request = new TestRequest(testbed, event.edits());
+ requests.add(request);
+ fakeClient.execute(request);
+ }
+ break;
+
+ case UPDATE_DELAYS:
+ break;
+
+ case CRASH_NODES:
+ break;
+
+ case START_NODES:
+ break;
+
+ default:
+ break;
+ }
+ if (testbed.checkLeaderCount()) {
+ Assert.assertTrue(testbed.getLeaderCount() < 2);
+ }
+ }
+ List<TestRequest> tmp = new ArrayList<TestRequest>();
+ for (TestRequest request : requests) {
+ switch(request.state()) {
+ case FAILED:
+ case DONE:
+ processedRequests.add(request);
+ break;
+ default:
+ tmp.add(request);
+ }
+ }
+ requests = tmp;
+ if (previousStateDumpTime + 2000L < System.currentTimeMillis()) {
+ long t1 = System.nanoTime();
+ testbed.dumpStates();
+ long t2 = System.nanoTime();
+ LOG.info("Dumping States took " + (t2-t1)/1000L + " us");
+ previousStateDumpTime = System.currentTimeMillis();
+ }
+ }
+
+ LOG.info(
+ "\n\n"
+ + "-------------------------------------------\n"
+ + "SHUTTING DOWN THE SIMULATOR ......\n"
+ + "-------------------------------------------\n"
+ );
+
+ fakeClient.shutdown();
+
+ try {
+ fakeClient.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+ } catch (InterruptedException ex) {
+ LOG.error("Time to die!");
+ System.exit(1);
+ }
+
+ int ndone = 0;
+ int nfail = 0;
+ for (TestRequest request : processedRequests) {
+ switch (request.state()) {
+ case FAILED:
+ nfail ++;
+ break;
+ case DONE:
+ ndone ++;
+ break;
+
+ default:
+ LOG.error("There are requests that have not been processed!");
+ System.exit(1);
+ }
+ }
+ LOG.info("\n\n"
+ + "--------------------------------------------------------\n"
+ + "Successfully committed " + ndone + " client requests\n"
+ + " Stats\n"
+ + " - droppped " + testbed.getPacketDropCount() + " packets (" + testbed.getHiccupPacketDropCount() + " due to hiccups)\n"
+ + "--------------------------------------------------------\n"
+ );
+ if (nfail > 0) {
+ LOG.error("There are failed requests!");
+ System.exit(1);
+ }
+
+ LOG.info("Clearing network states ......");
+ adversary.getChaos().updateNetworkStates(System.currentTimeMillis(), true);
+ LOG.info("-------- Verifying log consistency amongst all quorum members");
+ while (!RAFT_TEST_UTIL.verifyLogs(regionInfo.getQuorumInfo(), QUORUM_SIZE, true)) {
+ testbed.dumpStates();
+ if (testbed.checkLeaderCount()) {
+ Assert.assertTrue(testbed.getLeaderCount() < 2);
+ }
+ noThrowSleep(5000L);
+ }
+ LOG.info("-------- Done verifying log consistency amongst all quorum members");
+ testbed.dumpStates();
+ }
+ }
+
+
+ public void start() throws Exception {
+ start(new TestConfig());
+ }
+
+ public void start(TestConfig config) throws Exception {
+ PeerHandleAppendResponse.setGlobalMaxBatchLogs((int)config.getLong("max-batch-logs", 64));
+ int port = config.getInt("regionserver-port", -1);
+ if (port > 0) {
+ LOG.info("Setting the HBASE region server port to " + port);
+ RAFT_TEST_UTIL.getConf().setInt(HConstants.REGIONSERVER_PORT, port);
+ }
+
+ long timeout = config.getLong("handle-rpc-timeout", -1);
+ if (timeout > 0) {
+ LOG.info("Setting the RPC Error timeout to " + timeout + " ms");
+ RAFT_TEST_UTIL.getConf().setLong(HConstants.RAFT_PEERSERVER_HANDLE_RPC_TIMEOUT_MS, timeout);
+ }
+
+ RAFT_TEST_UTIL.setUsePeristentLog(usePersistentLog);
+ RAFT_TEST_UTIL.createRaftCluster(QUORUM_SIZE);
+ RAFT_TEST_UTIL.assertAllServersRunning();
+ regionInfo = RAFT_TEST_UTIL.initializePeers();
+ RAFT_TEST_UTIL.addQuorum(regionInfo, mockLogs);
+ RAFT_TEST_UTIL.startQuorum(regionInfo);
+ checker = new HealthChecker(regionInfo, RAFT_TEST_UTIL, autoRestartThriftServer, 30000L);
+ checker.start();
+ }
+
+ public void stop() throws Exception {
+ if (checker != null) {
+ checker.shutdown();
+ }
+ RAFT_TEST_UTIL.shutdown();
+ }
+
+ public LocalTestBed() {
+ this(null);
+ }
+
+ public LocalTestBed(List<int[]> logs) {
+ if (logs == null) {
+ logs = new ArrayList<int[]>();
+ for (int i=0; i<QUORUM_SIZE; i++) {
+ logs.add(new int[] {});
+ }
+ }
+ mockLogs = logs;
+ }
+
+ private static List<WALEdit> generateTestingWALEdit() {
+ KeyValue kv = KeyValue.createFirstOnRow(Bytes.toBytes("TestQuorum"));
+ return Arrays.asList(new WALEdit(Arrays.asList(kv)));
+ }
+
+ public void dumpStates() {
+ RAFT_TEST_UTIL.dumpStates(regionInfo);
+ LOG.info("Total Commit = " + commitSuccessCount.get()+ " successes and " + commitFailureCount.get() + " failures "
+ + " with " + getPacketDropCount() + "(" + getHiccupPacketDropCount() + ") total-dropped (hiccup) packets "
+ + " and " + RAFT_TEST_UTIL.getServerRestartCount() + " server restarts "
+ + " in " + (System.currentTimeMillis() - BigBang)/1000L + " secs");
+ }
+
+ public void incrementCommitCount(boolean success) {
+ if (success) {
+ commitSuccessCount.incrementAndGet();
+ } else {
+ commitFailureCount.incrementAndGet();
+ }
+ }
+
+ private boolean doCommit(final WALEdit edit) {
+ if (edit == null) {
+ return true;
+ }
+ retry_loop:
+ while (true) {
+ try {
+ QuorumAgent agent = getLeader();
+ if (agent == null) {
+ LOG.info("NULL leader. Sleep for some time ......");
+ try {
+ Thread.sleep(1000L);
+ } catch (InterruptedException ex) {
+ LOG.info("Time to Die!");
+ return false;
+ }
+ continue;
+ }
+ agent.syncAppend(edit);
+ // Verify all the logs across the majority are the same
+ RAFT_TEST_UTIL.verifyLogs(regionInfo.getQuorumInfo(), QUORUM_MAJORITY, false);
+ return true;
+ } catch (NewLeaderException e) {
+ LOG.warn("Got a new leader in the quorum: " + e.getNewLeaderAddress());
+ RAFT_TEST_UTIL.verifyLogs(regionInfo.getQuorumInfo(), QUORUM_MAJORITY, false);
+ } catch (Exception e) {
+ Throwable cause = e;
+ while (cause != null) {
+ if (cause instanceof NewLeaderException || cause instanceof LeaderNotReadyException) {
+ continue retry_loop;
+ }
+ cause = cause.getCause();
+ }
+ LOG.error("Errors: ", e);
+ }
+ }
+ }
+
+ public boolean testSingleCommit(List<WALEdit> edits) {
+ boolean success = true;
+ if (edits.size() == 0) {
+ return true;
+ }
+
+ for (final WALEdit edit : edits) {
+ if ((success = doCommit(edit)) == false) {
+ break;
+ }
+ }
+ return success;
+ }
+
+ public QuorumAgent getLeader() {
+ RaftQuorumContext leaderQuorum = null;
+ do {
+ int leaderCnt = 0;
+ for (LocalConsensusServer server : RAFT_TEST_UTIL.getServers().values()) {
+ RaftQuorumContext c = server.getHandler().getRaftQuorumContext(regionInfo.getEncodedName());
+ if (c != null && c.isLeader()) {
+ leaderQuorum = c;
+ leaderCnt++;
+ }
+ }
+ if (checkLeaderCount()) {
+ Assert.assertTrue(getLeaderCount() < 2);
+ }
+ } while (leaderQuorum == null );
+ return leaderQuorum.getQuorumAgentInstance();
+ }
+
+ public int getLeaderCount() {
+ int leaderCnt = 0;
+ for (LocalConsensusServer server : RAFT_TEST_UTIL.getServers().values()) {
+ RaftQuorumContext c = server.getHandler().getRaftQuorumContext(regionInfo.getEncodedName());
+ if (c != null && c.isLeader()) {
+ leaderCnt++;
+ }
+ }
+ return leaderCnt;
+ }
+
+
+
+ public static void noThrowSleep(long ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ public void runConsensusProtocol() throws Exception {
+ runConsensusProtocol(new TestConfig());
+ }
+
+ public void runConsensusProtocol(TestConfig config) throws Exception {
+ start(config);
+
+ try {
+ Chaos chaos = new Chaos(config);
+ Test test = new Test(config, this);
+ Adversary adversary = new Adversary(chaos, config);
+ test.init(adversary);
+ test.run();
+ LOG.info("Stopping the experiment ......");
+ stop();
+ } catch (Throwable fit) {
+ LOG.error("runConsensusProtocol caught a fit", fit);
+ System.exit(1);
+ }
+ }
+
+ public boolean checkLeaderCount() {
+ return checkLeaderCount;
+ }
+
+ public void parseOptions(String[] args) {
+ for (String arg : args) {
+ if (arg.equalsIgnoreCase("--check-leader-count")) {
+ checkLeaderCount = true;
+ } else if (arg.equalsIgnoreCase("--no-check-leader-count")) {
+ checkLeaderCount = false;
+ } else if (arg.equalsIgnoreCase("--persistent-log")) {
+ usePersistentLog = true;
+ } else if (arg.equalsIgnoreCase("--no-persistent-log")) {
+ usePersistentLog = false;
+ } else if (arg.equalsIgnoreCase("--auto-restart-thrift")) {
+ autoRestartThriftServer = true;
+ } else if (arg.equalsIgnoreCase("--no-auto-restart-thrift")) {
+ autoRestartThriftServer = false;
+ }
+ }
+ }
+
+ public long getPacketDropCount() {
+ return RAFT_TEST_UTIL.getPacketDropCount(regionInfo);
+ }
+
+ public long getHiccupPacketDropCount() {
+ return RAFT_TEST_UTIL.getHiccupPacketDropCount(regionInfo);
+ }
+
+ // shoplifted from http://preshing.com/20111007/how-to-generate-random-timings-for-a-poisson-process/
+ public static long poissonRandomInterarrivalDelay(Random prng, double lambda) {
+ return Math.max(0L, (long)(-Math.log(1.0 - prng.nextDouble()) * lambda));
+ }
+
+ public static void main(String[] args) throws Exception {
+ LocalTestBed testbed = new LocalTestBed();
+ testbed.parseOptions(args);
+ testbed.runConsensusProtocol(new TestConfig(args));
+ LOG.info("\n\n"
+ + "--------------------------------------------------------\n"
+ + "SUCCESS\n"
+ + " Stats\n"
+ + " - droppped " + testbed.getPacketDropCount() + " packets (" + testbed.getHiccupPacketDropCount() + " due to hiccups)\n"
+ + "--------------------------------------------------------\n"
+ );
+ System.exit(0);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/eca32aa4/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/RaftTestDataProvider.java
----------------------------------------------------------------------
diff --git a/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/RaftTestDataProvider.java b/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/RaftTestDataProvider.java
new file mode 100644
index 0000000..9efefb9
--- /dev/null
+++ b/hbase-consensus/src/test/java/org/apache/hadoop/hbase/consensus/RaftTestDataProvider.java
@@ -0,0 +1,130 @@
+package org.apache.hadoop.hbase.consensus;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class RaftTestDataProvider {
+
+ public static List<Object[]> getRaftBasicLogTestData() {
+ List<int[]> test1 = Arrays.asList(new int[][]{
+ {1},
+ {1},
+ {1},
+ {1},
+ {1}
+ });
+
+ List<int[]> test2 = Arrays.asList(new int[][] {
+ {1, 2},
+ {1},
+ {1},
+ {1},
+ {1}
+ });
+
+ List<int[]> test3 = Arrays.asList(new int[][] {
+ {1, 2},
+ {1, 2},
+ {1},
+ {1},
+ {1}
+ });
+
+ List<int[]> test4 = Arrays.asList(new int[][] {
+ {1, 2},
+ {1, 2},
+ {1, 2},
+ {1},
+ {1}
+ });
+
+
+ List<int[]> test5 = Arrays.asList(new int[][] {
+ {1, 2},
+ {1, 2},
+ {1, 2},
+ {1, 2},
+ {1}
+ });
+
+
+ List<int[]> test6 = Arrays.asList(new int[][] {
+ {1, 2},
+ {1, 2},
+ {1, 3},
+ {1, 3},
+ {1}
+ });
+
+ List<int[]> test7 = Arrays.asList(new int[][] {
+ {1, 2},
+ {1, 2},
+ {1, 3},
+ {1, 3},
+ {1, 3}
+ });
+
+ List<int[]> test8 = Arrays.asList(new int[][] {
+ {1, 1, 2},
+ {1, 1, 1},
+ {1, 1, 2},
+ {1, 1, 1},
+ {}
+ });
+
+ List<int[]> test9 = Arrays.asList(new int[][] {
+ {1, 1},
+ {1},
+ {1},
+ {},
+ {}
+ });
+
+ List<int[]> test10 = Arrays.asList(new int[][] {
+ {1, 1, 1, 4, 4, 5, 5, 6, 6, 6},
+ {1, 1, 1, 4},
+ {1, 1, 1, 4, 4, 5, 5, 6, 6, 6, 7},
+ {},
+ {}
+ });
+
+
+ Object[][] data = new Object[][] {
+ { test1 },
+ { test2 },
+ { test3 },
+ { test4 },
+ { test5 },
+ { test6 },
+ { test7 },
+ { test8 },
+ { test9 },
+ { test10 }
+ };
+ return Arrays.asList(data);
+ }
+
+ public static List<Object[]> getRaftBasicLogTestSeedData() {
+ List<int[]> test1 = Arrays.asList(new int[][]{
+ {1, 1, 1, 1, 1, 1 ,1},
+ {1, 1, 1},
+ {1, 1, 1, 1, 1, 1, 1},
+ {1, 1, 1, 1},
+ {1, 1, 1, 1, 1, 1, 1}
+ });
+
+ List<int[]> test2 = Arrays.asList(new int[][] {
+ {1, 1, 1, 1, 1, 1 ,1},
+ {1, 1, 1, 1, 1, 1 ,1},
+ {1, 1, 2},
+ {1},
+ {1, 1, 1, 1, 1, 1 ,1}
+ });
+
+ Object[][] data = new Object[][] {
+ { test1 },
+ { test2 },
+ };
+ return Arrays.asList(data);
+ }
+}