diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java
new file mode 100644
index 0000000..3c92800
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseClient.java
@@ -0,0 +1,144 @@
+package org.onlab.onos.store.service.impl;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import net.kuujo.copycat.protocol.Response.Status;
+import net.kuujo.copycat.protocol.SubmitRequest;
+import net.kuujo.copycat.protocol.SubmitResponse;
+import net.kuujo.copycat.spi.protocol.ProtocolClient;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.onlab.netty.Endpoint;
+import org.onlab.netty.NettyMessagingService;
+import org.onlab.onos.store.service.DatabaseException;
+import org.onlab.onos.store.service.ReadRequest;
+import org.onlab.onos.store.service.WriteRequest;
+
+public class DatabaseClient {
+
+    private final Endpoint copycatEp;
+    ProtocolClient client;
+    NettyMessagingService messagingService;
+
+    public DatabaseClient(Endpoint copycatEp) {
+        this.copycatEp = copycatEp;
+    }
+
+    private static String nextId() {
+        return UUID.randomUUID().toString();
+    }
+
+    public void activate() throws Exception {
+        messagingService = new NettyMessagingService(RandomUtils.nextInt(10000, 40000));
+        messagingService.activate();
+        client = new NettyProtocolClient(copycatEp, messagingService);
+    }
+
+    public void deactivate() throws Exception {
+        messagingService.deactivate();
+    }
+
+    public boolean createTable(String tableName) {
+
+        SubmitRequest request =
+                new SubmitRequest(
+                        nextId(),
+                        "createTable",
+                        Arrays.asList(tableName));
+        CompletableFuture<SubmitResponse> future = client.submit(request);
+        try {
+            return (boolean) future.get().result();
+        } catch (InterruptedException | ExecutionException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    public void dropTable(String tableName) {
+
+        SubmitRequest request =
+                new SubmitRequest(
+                        nextId(),
+                        "dropTable",
+                        Arrays.asList(tableName));
+        CompletableFuture<SubmitResponse> future = client.submit(request);
+        try {
+            if (future.get().status() == Status.OK) {
+                throw new DatabaseException(future.get().toString());
+            }
+
+        } catch (InterruptedException | ExecutionException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    public void dropAllTables() {
+
+        SubmitRequest request =
+                new SubmitRequest(
+                        nextId(),
+                        "dropAllTables",
+                        Arrays.asList());
+        CompletableFuture<SubmitResponse> future = client.submit(request);
+        try {
+            if (future.get().status() != Status.OK) {
+                throw new DatabaseException(future.get().toString());
+            }
+        } catch (InterruptedException | ExecutionException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public List<String> listTables() {
+
+        SubmitRequest request =
+                new SubmitRequest(
+                        nextId(),
+                        "listTables",
+                        Arrays.asList());
+        CompletableFuture<SubmitResponse> future = client.submit(request);
+        try {
+            return (List<String>) future.get().result();
+        } catch (InterruptedException | ExecutionException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public List<InternalReadResult> batchRead(List<ReadRequest> requests) {
+
+        SubmitRequest request = new SubmitRequest(
+                        nextId(),
+                        "read",
+                        Arrays.asList(requests));
+
+        CompletableFuture<SubmitResponse> future = client.submit(request);
+        try {
+            List<InternalReadResult> internalReadResults = (List<InternalReadResult>) future.get().result();
+            return internalReadResults;
+        } catch (InterruptedException | ExecutionException e) {
+            throw new DatabaseException(e);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public List<InternalWriteResult> batchWrite(List<WriteRequest> requests) {
+
+        SubmitRequest request = new SubmitRequest(
+                        nextId(),
+                        "write",
+                        Arrays.asList(requests));
+
+        CompletableFuture<SubmitResponse> future = client.submit(request);
+        try {
+            List<InternalWriteResult> internalWriteResults = (List<InternalWriteResult>) future.get().result();
+            return internalWriteResults;
+        } catch (InterruptedException | ExecutionException e) {
+            throw new DatabaseException(e);
+        }
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java
new file mode 100644
index 0000000..00ce12d
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseManager.java
@@ -0,0 +1,210 @@
+package org.onlab.onos.store.service.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import net.kuujo.copycat.Copycat;
+import net.kuujo.copycat.StateMachine;
+import net.kuujo.copycat.cluster.TcpCluster;
+import net.kuujo.copycat.cluster.TcpClusterConfig;
+import net.kuujo.copycat.cluster.TcpMember;
+import net.kuujo.copycat.log.ChronicleLog;
+import net.kuujo.copycat.log.Log;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.netty.Endpoint;
+import org.onlab.onos.cluster.ClusterService;
+import org.onlab.onos.cluster.ControllerNode;
+import org.onlab.onos.store.service.DatabaseAdminService;
+import org.onlab.onos.store.service.DatabaseException;
+import org.onlab.onos.store.service.DatabaseService;
+import org.onlab.onos.store.service.NoSuchTableException;
+import org.onlab.onos.store.service.OptimisticLockException;
+import org.onlab.onos.store.service.OptionalResult;
+import org.onlab.onos.store.service.PreconditionFailedException;
+import org.onlab.onos.store.service.ReadRequest;
+import org.onlab.onos.store.service.ReadResult;
+import org.onlab.onos.store.service.WriteAborted;
+import org.onlab.onos.store.service.WriteRequest;
+import org.onlab.onos.store.service.WriteResult;
+import org.slf4j.Logger;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Strongly consistent and durable state management service based on
+ * Copycat implementation of Raft consensus protocol.
+ */
+@Component(immediate = true)
+@Service
+public class DatabaseManager implements DatabaseService, DatabaseAdminService {
+
+    private final Logger log = getLogger(getClass());
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    ClusterService clusterService;
+
+    public static final String LOG_FILE_PREFIX = "onos-copy-cat-log";
+
+    private Copycat copycat;
+    private DatabaseClient client;
+
+    @Activate
+    public void activate() {
+        TcpMember localMember =
+                new TcpMember(
+                        clusterService.getLocalNode().ip().toString(),
+                        clusterService.getLocalNode().tcpPort());
+        List<TcpMember> remoteMembers = Lists.newArrayList();
+
+        for (ControllerNode node : clusterService.getNodes()) {
+            TcpMember member = new TcpMember(node.ip().toString(), node.tcpPort());
+            if (!member.equals(localMember)) {
+                remoteMembers.add(member);
+            }
+        }
+
+        // Configure the cluster.
+        TcpClusterConfig config = new TcpClusterConfig();
+
+        config.setLocalMember(localMember);
+        config.setRemoteMembers(remoteMembers.toArray(new TcpMember[]{}));
+
+        // Create the cluster.
+        TcpCluster cluster = new TcpCluster(config);
+
+        StateMachine stateMachine = new DatabaseStateMachine();
+        ControllerNode thisNode = clusterService.getLocalNode();
+        Log consensusLog = new ChronicleLog(LOG_FILE_PREFIX + "_" + thisNode.id());
+
+        copycat = new Copycat(stateMachine, consensusLog, cluster, new NettyProtocol());
+        copycat.start();
+
+        client = new DatabaseClient(new Endpoint(localMember.host(), localMember.port()));
+
+        log.info("Started.");
+    }
+
+    @Activate
+    public void deactivate() {
+        copycat.stop();
+    }
+
+    @Override
+    public boolean createTable(String name) {
+        return client.createTable(name);
+    }
+
+    @Override
+    public void dropTable(String name) {
+        client.dropTable(name);
+    }
+
+    @Override
+    public void dropAllTables() {
+        client.dropAllTables();
+    }
+
+    @Override
+    public List<String> listTables() {
+        return client.listTables();
+    }
+
+    @Override
+    public ReadResult read(ReadRequest request) {
+        return batchRead(Arrays.asList(request)).get(0).get();
+    }
+
+    @Override
+    public List<OptionalResult<ReadResult, DatabaseException>> batchRead(
+            List<ReadRequest> batch) {
+        List<OptionalResult<ReadResult, DatabaseException>> readResults = new ArrayList<>(batch.size());
+        for (InternalReadResult internalReadResult : client.batchRead(batch)) {
+            if (internalReadResult.status() == InternalReadResult.Status.NO_SUCH_TABLE) {
+                readResults.add(new DatabaseOperationResult<ReadResult, DatabaseException>(
+                        new NoSuchTableException()));
+            } else {
+                readResults.add(new DatabaseOperationResult<ReadResult, DatabaseException>(
+                        internalReadResult.result()));
+            }
+        }
+        return readResults;
+    }
+
+    @Override
+    public WriteResult write(WriteRequest request) {
+        return batchWrite(Arrays.asList(request)).get(0).get();
+    }
+
+    @Override
+    public List<OptionalResult<WriteResult, DatabaseException>> batchWrite(
+            List<WriteRequest> batch) {
+        List<OptionalResult<WriteResult, DatabaseException>> writeResults = new ArrayList<>(batch.size());
+        for (InternalWriteResult internalWriteResult : client.batchWrite(batch)) {
+            if (internalWriteResult.status() == InternalWriteResult.Status.NO_SUCH_TABLE) {
+                writeResults.add(new DatabaseOperationResult<WriteResult, DatabaseException>(
+                        new NoSuchTableException()));
+            } else if (internalWriteResult.status() == InternalWriteResult.Status.OPTIMISTIC_LOCK_FAILURE) {
+                writeResults.add(new DatabaseOperationResult<WriteResult, DatabaseException>(
+                        new OptimisticLockException()));
+            } else if (internalWriteResult.status() == InternalWriteResult.Status.PREVIOUS_VALUE_MISMATCH) {
+                // TODO: throw a different exception?
+                writeResults.add(new DatabaseOperationResult<WriteResult, DatabaseException>(
+                        new PreconditionFailedException()));
+            } else if (internalWriteResult.status() == InternalWriteResult.Status.ABORTED) {
+                writeResults.add(new DatabaseOperationResult<WriteResult, DatabaseException>(
+                        new WriteAborted()));
+            } else {
+                writeResults.add(new DatabaseOperationResult<WriteResult, DatabaseException>(
+                        internalWriteResult.result()));
+            }
+        }
+        return writeResults;
+
+    }
+
+    private class DatabaseOperationResult<R, E extends DatabaseException> implements OptionalResult<R, E> {
+
+        private final R result;
+        private final DatabaseException exception;
+
+        public DatabaseOperationResult(R result) {
+            this.result = result;
+            this.exception = null;
+        }
+
+        public DatabaseOperationResult(DatabaseException exception) {
+            this.result = null;
+            this.exception = exception;
+        }
+
+        @Override
+        public R get() {
+            if (result != null) {
+                return result;
+            }
+            throw exception;
+        }
+
+        @Override
+        public boolean hasValidResult() {
+            return result != null;
+        }
+
+        @Override
+        public String toString() {
+            if (result != null) {
+                return result.toString();
+            } else {
+                return exception.toString();
+            }
+        }
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseStateMachine.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseStateMachine.java
new file mode 100644
index 0000000..cbca729
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/DatabaseStateMachine.java
@@ -0,0 +1,169 @@
+package org.onlab.onos.store.service.impl;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import net.kuujo.copycat.Command;
+import net.kuujo.copycat.Query;
+import net.kuujo.copycat.StateMachine;
+
+import org.onlab.onos.store.serializers.KryoSerializer;
+import org.onlab.onos.store.service.ReadRequest;
+import org.onlab.onos.store.service.ReadResult;
+import org.onlab.onos.store.service.WriteRequest;
+import org.onlab.onos.store.service.WriteResult;
+import org.onlab.util.KryoNamespace;
+
+import com.google.common.collect.Maps;
+
+public class DatabaseStateMachine implements StateMachine {
+
+    public static final KryoSerializer SERIALIZER = new KryoSerializer() {
+        @Override
+        protected void setupKryoPool() {
+            serializerPool = KryoNamespace.newBuilder()
+                    .register(VersionedValue.class)
+                    .register(State.class)
+                    .register(NettyProtocol.COMMON)
+                    .build()
+                    .populate(1);
+        }
+    };
+
+    private State state = new State();
+
+    @Command
+    public boolean createTable(String tableName) {
+        return state.getTables().putIfAbsent(tableName, Maps.newHashMap()) == null;
+    }
+
+    @Command
+    public boolean dropTable(String tableName) {
+        return state.getTables().remove(tableName) != null;
+    }
+
+    @Command
+    public boolean dropAllTables() {
+        state.getTables().clear();
+        return true;
+    }
+
+    @Query
+    public Set<String> listTables() {
+        return state.getTables().keySet();
+    }
+
+    @Query
+    public List<InternalReadResult> read(List<ReadRequest> requests) {
+        List<InternalReadResult> results = new ArrayList<>(requests.size());
+        for (ReadRequest request : requests) {
+            Map<String, VersionedValue> table = state.getTables().get(request.tableName());
+            if (table == null) {
+                results.add(new InternalReadResult(InternalReadResult.Status.NO_SUCH_TABLE, null));
+                continue;
+            }
+            VersionedValue value = table.get(request.key());
+            results.add(new InternalReadResult(
+                    InternalReadResult.Status.OK,
+                    new ReadResult(
+                            request.tableName(),
+                            request.key(),
+                            value)));
+        }
+        return results;
+    }
+
+    @Command
+    public List<InternalWriteResult> write(List<WriteRequest> requests) {
+        boolean abort = false;
+        List<InternalWriteResult.Status> validationResults = new ArrayList<>(requests.size());
+        for (WriteRequest request : requests) {
+            Map<String, VersionedValue> table = state.getTables().get(request.tableName());
+            if (table == null) {
+                validationResults.add(InternalWriteResult.Status.NO_SUCH_TABLE);
+                abort = true;
+                continue;
+            }
+            VersionedValue value = table.get(request.key());
+            if (value == null) {
+                if (request.oldValue() != null) {
+                    validationResults.add(InternalWriteResult.Status.PREVIOUS_VALUE_MISMATCH);
+                    abort = true;
+                    continue;
+                } else if (request.previousVersion() >= 0) {
+                    validationResults.add(InternalWriteResult.Status.OPTIMISTIC_LOCK_FAILURE);
+                    abort = true;
+                    continue;
+                }
+            }
+            if (request.previousVersion() >= 0 && value.version() != request.previousVersion()) {
+                validationResults.add(InternalWriteResult.Status.OPTIMISTIC_LOCK_FAILURE);
+                abort = true;
+                continue;
+            }
+
+            validationResults.add(InternalWriteResult.Status.OK);
+        }
+
+        List<InternalWriteResult> results = new ArrayList<>(requests.size());
+
+        if (abort) {
+            for (InternalWriteResult.Status validationResult : validationResults) {
+                if (validationResult == InternalWriteResult.Status.OK) {
+                    results.add(new InternalWriteResult(InternalWriteResult.Status.ABORTED, null));
+                } else {
+                    results.add(new InternalWriteResult(validationResult, null));
+                }
+            }
+            return results;
+        }
+
+        for (WriteRequest request : requests) {
+            Map<String, VersionedValue> table = state.getTables().get(request.tableName());
+            synchronized (table) {
+                VersionedValue previousValue =
+                        table.put(request.key(), new VersionedValue(request.newValue(), state.nextVersion()));
+                results.add(new InternalWriteResult(
+                        InternalWriteResult.Status.OK,
+                        new WriteResult(request.tableName(), request.key(), previousValue)));
+            }
+        }
+        return results;
+    }
+
+    public class State {
+
+        private final Map<String, Map<String, VersionedValue>> tables =
+                Maps.newHashMap();
+        private long versionCounter = 1;
+
+        Map<String, Map<String, VersionedValue>> getTables() {
+            return tables;
+        }
+
+        long nextVersion() {
+            return versionCounter++;
+        }
+    }
+
+    @Override
+    public byte[] takeSnapshot() {
+        try {
+            return SERIALIZER.encode(state);
+        } catch (Exception e) {
+            e.printStackTrace();
+            return null;
+        }
+    }
+
+    @Override
+    public void installSnapshot(byte[] data) {
+        try {
+            this.state = SERIALIZER.decode(data);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/InternalReadResult.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/InternalReadResult.java
new file mode 100644
index 0000000..f6fcf51
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/InternalReadResult.java
@@ -0,0 +1,33 @@
+package org.onlab.onos.store.service.impl;
+
+import org.onlab.onos.store.service.ReadResult;
+
+public class InternalReadResult {
+
+    public enum Status {
+        OK,
+        NO_SUCH_TABLE
+    }
+
+    private final Status status;
+    private final ReadResult result;
+
+    public InternalReadResult(Status status, ReadResult result) {
+        this.status = status;
+        this.result = result;
+    }
+
+    public Status status() {
+        return status;
+    }
+
+    public ReadResult result() {
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return "InternalReadResult [status=" + status + ", result=" + result
+                + "]";
+    }
+}
\ No newline at end of file
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/InternalWriteResult.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/InternalWriteResult.java
new file mode 100644
index 0000000..e6dbb1f
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/InternalWriteResult.java
@@ -0,0 +1,30 @@
+package org.onlab.onos.store.service.impl;
+
+import org.onlab.onos.store.service.WriteResult;
+
+public class InternalWriteResult {
+
+    public enum Status {
+        OK,
+        ABORTED,
+        NO_SUCH_TABLE,
+        OPTIMISTIC_LOCK_FAILURE,
+        PREVIOUS_VALUE_MISMATCH
+    }
+
+    private final Status status;
+    private final WriteResult result;
+
+    public InternalWriteResult(Status status, WriteResult result) {
+        this.status = status;
+        this.result = result;
+    }
+
+    public Status status() {
+        return status;
+    }
+
+    public WriteResult result() {
+        return result;
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocol.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocol.java
new file mode 100644
index 0000000..9855ec6
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocol.java
@@ -0,0 +1,145 @@
+package org.onlab.onos.store.service.impl;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Vector;
+
+import net.kuujo.copycat.cluster.TcpClusterConfig;
+import net.kuujo.copycat.cluster.TcpMember;
+import net.kuujo.copycat.internal.log.ConfigurationEntry;
+import net.kuujo.copycat.internal.log.CopycatEntry;
+import net.kuujo.copycat.internal.log.OperationEntry;
+import net.kuujo.copycat.internal.log.SnapshotEntry;
+import net.kuujo.copycat.protocol.PingRequest;
+import net.kuujo.copycat.protocol.PingResponse;
+import net.kuujo.copycat.protocol.PollRequest;
+import net.kuujo.copycat.protocol.PollResponse;
+import net.kuujo.copycat.protocol.Response.Status;
+import net.kuujo.copycat.protocol.SubmitRequest;
+import net.kuujo.copycat.protocol.SubmitResponse;
+import net.kuujo.copycat.protocol.SyncRequest;
+import net.kuujo.copycat.protocol.SyncResponse;
+import net.kuujo.copycat.spi.protocol.Protocol;
+import net.kuujo.copycat.spi.protocol.ProtocolClient;
+import net.kuujo.copycat.spi.protocol.ProtocolServer;
+
+import org.onlab.onos.store.serializers.ImmutableListSerializer;
+import org.onlab.onos.store.serializers.ImmutableMapSerializer;
+import org.onlab.onos.store.serializers.ImmutableSetSerializer;
+import org.onlab.onos.store.serializers.KryoSerializer;
+import org.onlab.onos.store.service.ReadRequest;
+import org.onlab.onos.store.service.ReadResult;
+import org.onlab.onos.store.service.WriteRequest;
+import org.onlab.onos.store.service.WriteResult;
+import org.onlab.util.KryoNamespace;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.serializers.CollectionSerializer;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * {@link Protocol} based on {@link org.onlab.netty.NettyMessagingService}.
+ */
+public class NettyProtocol implements Protocol<TcpMember> {
+
+    public static final String COPYCAT_PING = "copycat-raft-consensus-ping";
+    public static final String COPYCAT_SYNC = "copycat-raft-consensus-sync";
+    public static final String COPYCAT_POLL = "copycat-raft-consensus-poll";
+    public static final String COPYCAT_SUBMIT = "copycat-raft-consensus-submit";
+
+    // TODO: make this configurable.
+    public static final long RETRY_INTERVAL_MILLIS = 2000;
+
+    private static final KryoNamespace COPYCAT = KryoNamespace.newBuilder()
+            .register(PingRequest.class)
+            .register(PingResponse.class)
+            .register(PollRequest.class)
+            .register(PollResponse.class)
+            .register(SyncRequest.class)
+            .register(SyncResponse.class)
+            .register(SubmitRequest.class)
+            .register(SubmitResponse.class)
+            .register(Status.class)
+            .register(ConfigurationEntry.class)
+            .register(SnapshotEntry.class)
+            .register(CopycatEntry.class)
+            .register(OperationEntry.class)
+            .register(TcpClusterConfig.class)
+            .register(TcpMember.class)
+            .build();
+
+    // TODO: Move to the right place.
+    private static final KryoNamespace CRAFT = KryoNamespace.newBuilder()
+            .register(ReadRequest.class)
+            .register(WriteRequest.class)
+            .register(InternalReadResult.class)
+            .register(InternalWriteResult.class)
+            .register(InternalReadResult.Status.class)
+            .register(WriteResult.class)
+            .register(ReadResult.class)
+            .register(InternalWriteResult.Status.class)
+            .register(VersionedValue.class)
+            .build();
+
+    public static final KryoNamespace COMMON = KryoNamespace.newBuilder()
+            .register(Arrays.asList().getClass(), new CollectionSerializer() {
+                @Override
+                @SuppressWarnings("rawtypes")
+                protected Collection<?> create(Kryo kryo, Input input, Class<Collection> type) {
+                    return new ArrayList();
+                }
+            })
+            .register(ImmutableMap.class, new ImmutableMapSerializer())
+            .register(ImmutableList.class, new ImmutableListSerializer())
+            .register(ImmutableSet.class, new ImmutableSetSerializer())
+            .register(
+                    Vector.class,
+                    ArrayList.class,
+                    Arrays.asList().getClass(),
+                    HashMap.class,
+                    HashSet.class,
+                    LinkedList.class,
+                    byte[].class)
+            .build();
+
+    public static final KryoSerializer SERIALIZER = new KryoSerializer() {
+        @Override
+        protected void setupKryoPool() {
+            serializerPool = KryoNamespace.newBuilder()
+                    .register(COPYCAT)
+                    .register(COMMON)
+                    .register(CRAFT)
+                    .build()
+                    .populate(1);
+        }
+    };
+
+    private NettyProtocolServer server = null;
+
+    // FIXME: This is a total hack.Assumes
+    // ProtocolServer is initialized before ProtocolClient
+    protected NettyProtocolServer getServer() {
+        if (server == null) {
+            throw new IllegalStateException("ProtocolServer is not initialized yet!");
+        }
+        return server;
+    }
+
+    @Override
+    public ProtocolServer createServer(TcpMember member) {
+        server = new NettyProtocolServer(member);
+        return server;
+    }
+
+    @Override
+    public ProtocolClient createClient(TcpMember member) {
+        return new NettyProtocolClient(this, member);
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocolClient.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocolClient.java
new file mode 100644
index 0000000..a791990
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocolClient.java
@@ -0,0 +1,148 @@
+package org.onlab.onos.store.service.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import net.kuujo.copycat.cluster.TcpMember;
+import net.kuujo.copycat.protocol.PingRequest;
+import net.kuujo.copycat.protocol.PingResponse;
+import net.kuujo.copycat.protocol.PollRequest;
+import net.kuujo.copycat.protocol.PollResponse;
+import net.kuujo.copycat.protocol.SubmitRequest;
+import net.kuujo.copycat.protocol.SubmitResponse;
+import net.kuujo.copycat.protocol.SyncRequest;
+import net.kuujo.copycat.protocol.SyncResponse;
+import net.kuujo.copycat.spi.protocol.ProtocolClient;
+
+import org.onlab.netty.Endpoint;
+import org.onlab.netty.NettyMessagingService;
+import org.slf4j.Logger;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * {@link NettyMessagingService} based Copycat protocol client.
+ */
+public class NettyProtocolClient implements ProtocolClient {
+
+    private final Logger log = getLogger(getClass());
+    private static final ThreadFactory THREAD_FACTORY =
+            new ThreadFactoryBuilder().setNameFormat("copycat-netty-messaging-%d").build();
+
+    // Remote endpoint, this client instance is used
+    // for communicating with.
+    private final Endpoint remoteEp;
+    private final NettyMessagingService messagingService;
+
+    // TODO: Is 10 the right number of threads?
+    private static final ScheduledExecutorService THREAD_POOL =
+            new ScheduledThreadPoolExecutor(10, THREAD_FACTORY);
+
+    public NettyProtocolClient(NettyProtocol protocol, TcpMember member) {
+        this(new Endpoint(member.host(), member.port()), protocol.getServer().getNettyMessagingService());
+    }
+
+    public NettyProtocolClient(Endpoint remoteEp, NettyMessagingService messagingService) {
+        this.remoteEp = remoteEp;
+        this.messagingService = messagingService;
+    }
+
+    @Override
+    public CompletableFuture<PingResponse> ping(PingRequest request) {
+        return requestReply(request);
+    }
+
+    @Override
+    public CompletableFuture<SyncResponse> sync(SyncRequest request) {
+        return requestReply(request);
+    }
+
+    @Override
+    public CompletableFuture<PollResponse> poll(PollRequest request) {
+        return requestReply(request);
+    }
+
+    @Override
+    public CompletableFuture<SubmitResponse> submit(SubmitRequest request) {
+        return requestReply(request);
+    }
+
+    @Override
+    public CompletableFuture<Void> connect() {
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public CompletableFuture<Void> close() {
+        return CompletableFuture.completedFuture(null);
+    }
+
+    public <I> String messageType(I input) {
+        Class<?> clazz = input.getClass();
+        if (clazz.equals(PollRequest.class)) {
+            return NettyProtocol.COPYCAT_POLL;
+        } else if (clazz.equals(SyncRequest.class)) {
+            return NettyProtocol.COPYCAT_SYNC;
+        } else if (clazz.equals(SubmitRequest.class)) {
+            return NettyProtocol.COPYCAT_SUBMIT;
+        } else if (clazz.equals(PingRequest.class)) {
+            return NettyProtocol.COPYCAT_PING;
+        } else {
+            throw new IllegalArgumentException("Unknown class " + clazz.getName());
+        }
+
+    }
+
+    private <I, O> CompletableFuture<O> requestReply(I request) {
+        CompletableFuture<O> future = new CompletableFuture<>();
+        THREAD_POOL.schedule(new RPCTask<I, O>(request, future), 0, TimeUnit.MILLISECONDS);
+        return future;
+    }
+
+    private class RPCTask<I, O> implements Runnable {
+
+        private final String messageType;
+        private final byte[] payload;
+
+        private final CompletableFuture<O> future;
+
+        public RPCTask(I request, CompletableFuture<O> future) {
+            this.messageType = messageType(request);
+            this.payload = NettyProtocol.SERIALIZER.encode(request);
+            this.future = future;
+        }
+
+        @Override
+        public void run() {
+            try {
+                byte[] response = messagingService
+                    .sendAndReceive(remoteEp, messageType, payload)
+                    .get(NettyProtocol.RETRY_INTERVAL_MILLIS, TimeUnit.MILLISECONDS);
+                future.complete(NettyProtocol.SERIALIZER.decode(response));
+
+            } catch (IOException | InterruptedException | ExecutionException | TimeoutException e) {
+                if (messageType.equals(NettyProtocol.COPYCAT_SYNC) ||
+                        messageType.equals(NettyProtocol.COPYCAT_PING)) {
+                    log.warn("Request to {} failed. Will retry "
+                            + "in {} ms", remoteEp, NettyProtocol.RETRY_INTERVAL_MILLIS);
+                    THREAD_POOL.schedule(
+                            this,
+                            NettyProtocol.RETRY_INTERVAL_MILLIS,
+                            TimeUnit.MILLISECONDS);
+                } else {
+                    future.completeExceptionally(e);
+                }
+            } catch (Exception e) {
+                future.completeExceptionally(e);
+            }
+        }
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocolServer.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocolServer.java
new file mode 100644
index 0000000..d06999e
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/NettyProtocolServer.java
@@ -0,0 +1,115 @@
+package org.onlab.onos.store.service.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+import net.kuujo.copycat.cluster.TcpMember;
+import net.kuujo.copycat.protocol.PingRequest;
+import net.kuujo.copycat.protocol.PollRequest;
+import net.kuujo.copycat.protocol.RequestHandler;
+import net.kuujo.copycat.protocol.SubmitRequest;
+import net.kuujo.copycat.protocol.SyncRequest;
+import net.kuujo.copycat.spi.protocol.ProtocolServer;
+
+import org.onlab.netty.Message;
+import org.onlab.netty.MessageHandler;
+import org.onlab.netty.NettyMessagingService;
+import org.slf4j.Logger;
+
+/**
+ * {@link NettyMessagingService} based Copycat protocol server.
+ */
+public class NettyProtocolServer implements ProtocolServer {
+
+    private final Logger log = getLogger(getClass());
+
+    private final NettyMessagingService messagingService;
+    private RequestHandler handler;
+
+
+    public NettyProtocolServer(TcpMember member) {
+        messagingService = new NettyMessagingService(member.host(), member.port());
+
+        messagingService.registerHandler(NettyProtocol.COPYCAT_PING, new CopycatMessageHandler<PingRequest>());
+        messagingService.registerHandler(NettyProtocol.COPYCAT_SYNC, new CopycatMessageHandler<SyncRequest>());
+        messagingService.registerHandler(NettyProtocol.COPYCAT_POLL, new CopycatMessageHandler<PollRequest>());
+        messagingService.registerHandler(NettyProtocol.COPYCAT_SUBMIT, new CopycatMessageHandler<SubmitRequest>());
+    }
+
+    protected NettyMessagingService getNettyMessagingService() {
+        return messagingService;
+    }
+
+    @Override
+    public void requestHandler(RequestHandler handler) {
+        this.handler = handler;
+    }
+
+    @Override
+    public CompletableFuture<Void> listen() {
+        try {
+            messagingService.activate();
+            return CompletableFuture.completedFuture(null);
+        } catch (Exception e) {
+            CompletableFuture<Void> future = new CompletableFuture<>();
+            future.completeExceptionally(e);
+            return future;
+        }
+    }
+
+    @Override
+    public CompletableFuture<Void> close() {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        try {
+            messagingService.deactivate();
+            future.complete(null);
+            return future;
+        } catch (Exception e) {
+            future.completeExceptionally(e);
+            return future;
+        }
+    }
+
+    private class CopycatMessageHandler<T> implements MessageHandler {
+
+        @Override
+        public void handle(Message message) throws IOException {
+            T request = NettyProtocol.SERIALIZER.decode(message.payload());
+            if (request.getClass().equals(PingRequest.class)) {
+                handler.ping((PingRequest) request).whenComplete((response, error) -> {
+                    try {
+                        message.respond(NettyProtocol.SERIALIZER.encode(response));
+                    } catch (Exception e) {
+                        log.error("Failed to respond to ping request", e);
+                    }
+                });
+            } else if (request.getClass().equals(PollRequest.class)) {
+                handler.poll((PollRequest) request).whenComplete((response, error) -> {
+                    try {
+                        message.respond(NettyProtocol.SERIALIZER.encode(response));
+                    } catch (Exception e) {
+                        log.error("Failed to respond to poll request", e);
+                    }
+                });
+            } else if (request.getClass().equals(SyncRequest.class)) {
+                handler.sync((SyncRequest) request).whenComplete((response, error) -> {
+                    try {
+                        message.respond(NettyProtocol.SERIALIZER.encode(response));
+                    } catch (Exception e) {
+                        log.error("Failed to respond to sync request", e);
+                    }
+                });
+            } else if (request.getClass().equals(SubmitRequest.class)) {
+                handler.submit((SubmitRequest) request).whenComplete((response, error) -> {
+                    try {
+                        message.respond(NettyProtocol.SERIALIZER.encode(response));
+                    } catch (Exception e) {
+                        log.error("Failed to respond to submit request", e);
+                    }
+                });
+            }
+        }
+    }
+}
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/VersionedValue.java b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/VersionedValue.java
new file mode 100644
index 0000000..31bdcc2
--- /dev/null
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/service/impl/VersionedValue.java
@@ -0,0 +1,44 @@
+package org.onlab.onos.store.service.impl;
+
+import java.util.Arrays;
+
+/**
+ * Wrapper object that holds the object (as byte array) and its version.
+ */
+public class VersionedValue {
+
+    private final byte[] value;
+    private final long version;
+
+    /**
+     * Creates a new instance with the specified value and version.
+     * @param value
+     * @param version
+     */
+    public VersionedValue(byte[] value, long version) {
+        this.value = value;
+        this.version = version;
+    }
+
+    /**
+     * Returns the value.
+     * @return value.
+     */
+    public byte[] value() {
+        return value;
+    }
+
+    /**
+     * Returns the version.
+     * @return version.
+     */
+    public long version() {
+        return version;
+    }
+
+    @Override
+    public String toString() {
+        return "VersionedValue [value=" + Arrays.toString(value) + ", version="
+                + version + "]";
+    }
+}