bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [12/49] bookkeeper git commit: BOOKKEEPER-769: Remove the Hedwig Code
Date Wed, 16 Mar 2016 03:44:22 GMT
http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/LocalDBPersistenceManager.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/LocalDBPersistenceManager.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/LocalDBPersistenceManager.java
deleted file mode 100644
index b3c5dc9..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/LocalDBPersistenceManager.java
+++ /dev/null
@@ -1,491 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import java.io.File;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-import java.math.BigInteger;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-
-import javax.sql.rowset.serial.SerialBlob;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static com.google.common.base.Charsets.UTF_8;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
-import org.apache.hedwig.exceptions.PubSubException.UnexpectedConditionException;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-import org.apache.hedwig.protoextensions.MessageIdUtils;
-import org.apache.hedwig.server.persistence.ScanCallback.ReasonForFinish;
-import org.apache.hedwig.util.Callback;
-import org.apache.hedwig.util.FileUtils;
-
-public class LocalDBPersistenceManager implements PersistenceManagerWithRangeScan {
-    private static final Logger logger = LoggerFactory.getLogger(LocalDBPersistenceManager.class);
-
-    static String connectionURL;
-
-    static {
-        try {
-            File tempDir = FileUtils.createTempDirectory("derby", null);
-
-            // Since derby needs to create it, I will have to delete it first
-            if (!tempDir.delete()) {
-                throw new IOException("Could not delete dir: " + tempDir.getAbsolutePath());
-            }
-            connectionURL = "jdbc:derby:" + tempDir.getAbsolutePath() + ";create=true";
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-
-    }
-
-    private static final ThreadLocal<Connection> threadLocalConnection = new ThreadLocal<Connection>() {
-        @Override
-        protected Connection initialValue() {
-            try {
-                return DriverManager.getConnection(connectionURL);
-            } catch (SQLException e) {
-                logger.error("Could not connect to derby", e);
-                return null;
-            }
-        }
-    };
-
-    private static final ThreadLocal<MessageDigest> threadLocalDigest = new ThreadLocal<MessageDigest>() {
-        @Override
-        protected MessageDigest initialValue() {
-            try {
-                return MessageDigest.getInstance("MD5");
-            } catch (NoSuchAlgorithmException e) {
-                logger.error("Could not find MD5 hash", e);
-                return null;
-            }
-        }
-    };
-    static final String ID_FIELD_NAME = "id";
-    static final String MSG_FIELD_NAME = "msg";
-    static final String driver = "org.apache.derby.jdbc.EmbeddedDriver";
-
-    static final int SCAN_CHUNK = 1000;
-
-    /**
-     * Having trouble restarting the database multiple times from within the
-     * same jvm. Hence to facilitate units tests, we are just going to have a
-     * version number that we will append to every table name. This version
-     * number will be incremented in lieu of shutting down the database and
-     * restarting it, so that we get different table names, and it behaves like
-     * a brand new database
-     */
-    private int version = 0;
-
-    ConcurrentMap<ByteString, MessageSeqId> currTopicSeqIds = new ConcurrentHashMap<ByteString, MessageSeqId>();
-
-    static LocalDBPersistenceManager instance = new LocalDBPersistenceManager();
-
-    public static LocalDBPersistenceManager instance() {
-        return instance;
-    }
-
-    private LocalDBPersistenceManager() {
-
-        try {
-            Class.forName(driver).newInstance();
-            logger.info("Derby Driver loaded");
-        } catch (java.lang.ClassNotFoundException e) {
-            logger.error("Derby driver not found", e);
-        } catch (InstantiationException e) {
-            logger.error("Could not instantiate derby driver", e);
-        } catch (IllegalAccessException e) {
-            logger.error("Could not instantiate derby driver", e);
-        }
-    }
-
-    @Override
-    public void stop() {
-        // do nothing
-    }
-
-    /**
-     * Ensures that at least the default seq-id exists in the map for the given
-     * topic. Checks for race conditions (.e.g, another thread inserts the
-     * default id before us), and returns the latest seq-id value in the map
-     *
-     * @param topic
-     * @return
-     */
-    private MessageSeqId ensureSeqIdExistsForTopic(ByteString topic) {
-        MessageSeqId presentSeqIdInMap = currTopicSeqIds.get(topic);
-
-        if (presentSeqIdInMap != null) {
-            return presentSeqIdInMap;
-        }
-
-        presentSeqIdInMap = MessageSeqId.newBuilder().setLocalComponent(0).build();
-        MessageSeqId oldSeqIdInMap = currTopicSeqIds.putIfAbsent(topic, presentSeqIdInMap);
-
-        if (oldSeqIdInMap != null) {
-            return oldSeqIdInMap;
-        }
-        return presentSeqIdInMap;
-
-    }
-
-    /**
-     * Adjust the current seq id of the topic based on the message we are about
-     * to publish. The local component of the current seq-id is always
-     * incremented by 1. For the other components, there are two cases:
-     *
-     * 1. If the message to be published doesn't have a seq-id (locally
-     * published messages), the other components are left as is.
-     *
-     * 2. If the message to be published has a seq-id, we take the max of the
-     * current one we have, and that in the message to be published.
-     *
-     * @param topic
-     * @param messageToPublish
-     * @return The value of the local seq-id obtained after incrementing the
-     *         local component. This value should be used as an id while
-     *         persisting to Derby
-     * @throws UnexpectedConditionException
-     */
-    private long adjustTopicSeqIdForPublish(ByteString topic, Message messageToPublish)
-            throws UnexpectedConditionException {
-        long retValue = 0;
-        MessageSeqId oldId;
-        MessageSeqId.Builder newIdBuilder = MessageSeqId.newBuilder();
-
-        do {
-            oldId = ensureSeqIdExistsForTopic(topic);
-
-            // Increment our own component by 1
-            retValue = oldId.getLocalComponent() + 1;
-            newIdBuilder.setLocalComponent(retValue);
-
-            if (messageToPublish.hasMsgId()) {
-                // take a region-wise max
-                MessageIdUtils.takeRegionMaximum(newIdBuilder, messageToPublish.getMsgId(), oldId);
-
-            } else {
-                newIdBuilder.addAllRemoteComponents(oldId.getRemoteComponentsList());
-            }
-        } while (!currTopicSeqIds.replace(topic, oldId, newIdBuilder.build()));
-
-        return retValue;
-
-    }
-
-    public long getSeqIdAfterSkipping(ByteString topic, long seqId, int skipAmount) {
-        return seqId + skipAmount;
-    }
-
-    public void persistMessage(PersistRequest request) {
-
-        Connection conn = threadLocalConnection.get();
-
-        Callback<MessageSeqId> callback = request.getCallback();
-        Object ctx = request.getCtx();
-        ByteString topic = request.getTopic();
-        Message message = request.getMessage();
-
-        if (conn == null) {
-            callback.operationFailed(ctx, new ServiceDownException("Not connected to derby"));
-            return;
-        }
-
-        long seqId;
-
-        try {
-            seqId = adjustTopicSeqIdForPublish(topic, message);
-        } catch (UnexpectedConditionException e) {
-            callback.operationFailed(ctx, e);
-            return;
-        }
-        PreparedStatement stmt;
-
-        boolean triedCreatingTable = false;
-        while (true) {
-            try {
-                message.getBody();
-                stmt = conn.prepareStatement("INSERT INTO " + getTableNameForTopic(topic) + " VALUES(?,?)");
-                stmt.setLong(1, seqId);
-                stmt.setBlob(2, new SerialBlob(message.toByteArray()));
-
-                int rowCount = stmt.executeUpdate();
-                stmt.close();
-                if (rowCount != 1) {
-                    logger.error("Unexpected number of affected rows from derby");
-                    callback.operationFailed(ctx, new ServiceDownException("Unexpected response from derby"));
-                    return;
-                }
-                break;
-            } catch (SQLException sqle) {
-                String theError = (sqle).getSQLState();
-                if (theError.equals("42X05") && !triedCreatingTable) {
-                    createTable(conn, topic);
-                    triedCreatingTable = true;
-                    continue;
-                }
-
-                logger.error("Error while executing derby insert", sqle);
-                callback.operationFailed(ctx, new ServiceDownException(sqle));
-                return;
-            }
-        }
-        callback.operationFinished(ctx, MessageIdUtils.mergeLocalSeqId(message, seqId).getMsgId());
-    }
-
-    /*
-     * This method does not throw an exception because another thread might
-     * sneak in and create the table before us
-     */
-    private void createTable(Connection conn, ByteString topic) {
-        Statement stmt = null;
-        try {
-            stmt = conn.createStatement();
-            String tableName = getTableNameForTopic(topic);
-            stmt.execute("CREATE TABLE " + tableName + " (" + ID_FIELD_NAME + " BIGINT NOT NULL CONSTRAINT ID_PK_"
-                    + tableName + " PRIMARY KEY," + MSG_FIELD_NAME + " BLOB(2M) NOT NULL)");
-        } catch (SQLException e) {
-            logger.debug("Could not create table", e);
-        } finally {
-            try {
-                if (stmt != null) {
-                    stmt.close();
-                }
-            } catch (SQLException e) {
-                logger.error("Error closing statement", e);
-            }
-        }
-    }
-
-    public MessageSeqId getCurrentSeqIdForTopic(ByteString topic) {
-        return ensureSeqIdExistsForTopic(topic);
-    }
-
-    public void scanSingleMessage(ScanRequest request) {
-        scanMessagesInternal(request.getTopic(), request.getStartSeqId(), 1, Long.MAX_VALUE, request.getCallback(),
-                             request.getCtx(), 1);
-        return;
-    }
-
-    public void scanMessages(RangeScanRequest request) {
-        scanMessagesInternal(request.getTopic(), request.getStartSeqId(), request.getMessageLimit(), request
-                             .getSizeLimit(), request.getCallback(), request.getCtx(), SCAN_CHUNK);
-        return;
-    }
-
-    private String getTableNameForTopic(ByteString topic) {
-        String src = (topic.toStringUtf8() + "_" + version);
-        threadLocalDigest.get().reset();
-        byte[] digest = threadLocalDigest.get().digest(src.getBytes(UTF_8));
-        BigInteger bigInt = new BigInteger(1,digest);
-        return String.format("TABLE_%032X", bigInt);
-    }
-
-    private void scanMessagesInternal(ByteString topic, long startSeqId, int messageLimit, long sizeLimit,
-                                      ScanCallback callback, Object ctx, int scanChunk) {
-
-        Connection conn = threadLocalConnection.get();
-
-        if (conn == null) {
-            callback.scanFailed(ctx, new ServiceDownException("Not connected to derby"));
-            return;
-        }
-
-        long currentSeqId;
-        currentSeqId = startSeqId;
-
-        PreparedStatement stmt = null;
-        try {
-            try {
-                stmt = conn.prepareStatement("SELECT * FROM " + getTableNameForTopic(topic) + " WHERE " + ID_FIELD_NAME
-                                             + " >= ?  AND " + ID_FIELD_NAME + " <= ?");
-
-            } catch (SQLException sqle) {
-                String theError = (sqle).getSQLState();
-                if (theError.equals("42X05")) {
-                    // No table, scan is over
-                    callback.scanFinished(ctx, ReasonForFinish.NO_MORE_MESSAGES);
-                    return;
-                } else {
-                    throw sqle;
-                }
-            }
-
-            int numMessages = 0;
-            long totalSize = 0;
-
-            while (true) {
-
-                stmt.setLong(1, currentSeqId);
-                stmt.setLong(2, currentSeqId + scanChunk);
-
-                if (!stmt.execute()) {
-                    String errorMsg = "Select query did not return a result set";
-                    logger.error(errorMsg);
-                    stmt.close();
-                    callback.scanFailed(ctx, new ServiceDownException(errorMsg));
-                    return;
-                }
-
-                ResultSet resultSet = stmt.getResultSet();
-
-                if (!resultSet.next()) {
-                    stmt.close();
-                    callback.scanFinished(ctx, ReasonForFinish.NO_MORE_MESSAGES);
-                    return;
-                }
-
-                do {
-
-                    long localSeqId = resultSet.getLong(1);
-
-                    Message.Builder messageBuilder = Message.newBuilder().mergeFrom(resultSet.getBinaryStream(2));
-
-                    // Merge in the local seq-id since that is not stored with
-                    // the message
-                    Message message = MessageIdUtils.mergeLocalSeqId(messageBuilder, localSeqId);
-
-                    callback.messageScanned(ctx, message);
-                    numMessages++;
-                    totalSize += message.getBody().size();
-
-                    if (numMessages > messageLimit) {
-                        stmt.close();
-                        callback.scanFinished(ctx, ReasonForFinish.NUM_MESSAGES_LIMIT_EXCEEDED);
-                        return;
-                    } else if (totalSize > sizeLimit) {
-                        stmt.close();
-                        callback.scanFinished(ctx, ReasonForFinish.SIZE_LIMIT_EXCEEDED);
-                        return;
-                    }
-
-                } while (resultSet.next());
-
-                currentSeqId += SCAN_CHUNK;
-            }
-        } catch (SQLException e) {
-            logger.error("SQL Exception", e);
-            callback.scanFailed(ctx, new ServiceDownException(e));
-            return;
-        } catch (IOException e) {
-            logger.error("Message stored in derby is not parseable", e);
-            callback.scanFailed(ctx, new ServiceDownException(e));
-            return;
-        } finally {
-            try {
-                if (stmt != null) {
-                    stmt.close();
-                }
-            } catch (SQLException e) {
-                logger.error("Error closing statement", e);
-            }
-        }
-    }
-
-    public void deliveredUntil(ByteString topic, Long seqId) {
-        // noop
-    }
-
-    public void consumedUntil(ByteString topic, Long seqId) {
-        Connection conn = threadLocalConnection.get();
-        if (conn == null) {
-            logger.error("Not connected to derby");
-            return;
-        }
-        PreparedStatement stmt = null;
-        try {
-            stmt = conn.prepareStatement("DELETE FROM " + getTableNameForTopic(topic) + " WHERE " + ID_FIELD_NAME
-                                         + " <= ?");
-            stmt.setLong(1, seqId);
-            int rowCount = stmt.executeUpdate();
-            if (logger.isDebugEnabled()) {
-              logger.debug("Deleted " + rowCount + " records for topic: " + topic.toStringUtf8()
-                  + ", seqId: " + seqId);
-            }
-        } catch (SQLException sqle) {
-            String theError = (sqle).getSQLState();
-            if (theError.equals("42X05")) {
-                logger.warn("Table for topic (" + topic + ") does not exist so no consumed messages to delete!");
-            } else
-                logger.error("Error while executing derby delete for consumed messages", sqle);
-        } finally {
-            try {
-                if (stmt != null) {
-                    stmt.close();
-                }
-            } catch (SQLException e) {
-                logger.error("Error closing statement", e);
-            }
-        }
-    }
-
-    public void setMessageBound(ByteString topic, Integer bound) {
-        // noop; Maybe implement later
-    }
-
-    public void clearMessageBound(ByteString topic) {
-        // noop; Maybe implement later
-    }
-
-    public void consumeToBound(ByteString topic) {
-        // noop; Maybe implement later
-    }
-
-    @Override
-    protected void finalize() throws Throwable {
-        if (driver.equals("org.apache.derby.jdbc.EmbeddedDriver")) {
-            boolean gotSQLExc = false;
-            // This is weird: on normal shutdown, it throws an exception
-            try {
-                DriverManager.getConnection("jdbc:derby:;shutdown=true").close();
-            } catch (SQLException se) {
-                if (se.getSQLState().equals("XJ015")) {
-                    gotSQLExc = true;
-                }
-            }
-            if (!gotSQLExc) {
-                logger.error("Database did not shut down normally");
-            } else {
-                logger.info("Database shut down normally");
-            }
-        }
-        super.finalize();
-    }
-
-    public void reset() {
-        // just move the namespace over to the next one
-        version++;
-        currTopicSeqIds.clear();
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/MapMethods.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/MapMethods.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/MapMethods.java
deleted file mode 100644
index f640723..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/MapMethods.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import java.util.Collection;
-import java.util.Map;
-
-public class MapMethods {
-
-    public static <K, V> V getAfterInsertingIfAbsent(Map<K, V> map, K key, Factory<V> valueFactory) {
-        V value = map.get(key);
-
-        if (value == null) {
-            value = valueFactory.newInstance();
-            map.put(key, value);
-        }
-
-        return value;
-    }
-
-    public static <K, V, Z extends Collection<V>> void addToMultiMap(Map<K, Z> map, K key, V value,
-            Factory<Z> valueFactory) {
-        Collection<V> collection = getAfterInsertingIfAbsent(map, key, valueFactory);
-
-        collection.add(value);
-
-    }
-
-    public static <K, V, Z extends Collection<V>> boolean removeFromMultiMap(Map<K, Z> map, K key, V value) {
-        Collection<V> collection = map.get(key);
-
-        if (collection == null) {
-            return false;
-        }
-
-        if (!collection.remove(value)) {
-            return false;
-        } else {
-            if (collection.isEmpty()) {
-                map.remove(key);
-            }
-            return true;
-        }
-
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistRequest.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistRequest.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistRequest.java
deleted file mode 100644
index d137fe6..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistRequest.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.protocol.PubSubProtocol;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.util.Callback;
-
-/**
- * Encapsulates a request to persist a given message on a given topic. The
- * request is completed asynchronously, callback and context are provided
- *
- */
-public class PersistRequest {
-    ByteString topic;
-    Message message;
-    private Callback<PubSubProtocol.MessageSeqId> callback;
-    Object ctx;
-
-    public PersistRequest(ByteString topic, Message message, Callback<PubSubProtocol.MessageSeqId> callback, Object ctx) {
-        this.topic = topic;
-        this.message = message;
-        this.callback = callback;
-        this.ctx = ctx;
-    }
-
-    public ByteString getTopic() {
-        return topic;
-    }
-
-    public Message getMessage() {
-        return message;
-    }
-
-    public Callback<PubSubProtocol.MessageSeqId> getCallback() {
-        return callback;
-    }
-
-    public Object getCtx() {
-        return ctx;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManager.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManager.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManager.java
deleted file mode 100644
index a295fc7..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManager.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.exceptions.PubSubException.ServerNotResponsibleForTopicException;
-import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-
-/**
- * An implementation of this interface will persist messages in order and assign
- * a seqId to each persisted message. SeqId need not be a single number in
- * general. SeqId is opaque to all layers above {@link PersistenceManager}. Only
- * the {@link PersistenceManager} needs to understand the format of the seqId
- * and maintain it in such a way that there is a total order on the seqIds of a
- * topic.
- *
- */
-public interface PersistenceManager {
-
-    /**
-     * Executes the given persist request asynchronously. When done, the
-     * callback specified in the request object is called with the result of the
-     * operation set to the {@link LocalMessageSeqId} assigned to the persisted
-     * message.
-     */
-    public void persistMessage(PersistRequest request);
-
-    /**
-     * Get the seqId of the last message that has been persisted to the given
-     * topic. The returned seqId will be set as the consume position of any
-     * brand new subscription on this topic.
-     *
-     * Note that the return value may quickly become invalid because a
-     * {@link #persistMessage(String, PublishedMessage)} call from another
-     * thread succeeds. For us, the typical use case is choosing the consume
-     * position of a new subscriber. Since the subscriber need not receive all
-     * messages that are published while the subscribe call is in progress, such
-     * loose semantics from this method is acceptable.
-     *
-     * @param topic
-     * @return the seqId of the last persisted message.
-     * @throws ServerNotResponsibleForTopicException
-     */
-    public MessageSeqId getCurrentSeqIdForTopic(ByteString topic) throws ServerNotResponsibleForTopicException;
-
-    /**
-     * Executes the given scan request
-     *
-     */
-    public void scanSingleMessage(ScanRequest request);
-
-    /**
-     * Gets the next seq-id. This method should never block.
-     */
-    public long getSeqIdAfterSkipping(ByteString topic, long seqId, int skipAmount);
-
-    /**
-     * Hint that the messages until the given seqId have been delivered and wont
-     * be needed unless there is a failure of some kind
-     */
-    public void deliveredUntil(ByteString topic, Long seqId);
-
-    /**
-     * Hint that the messages until the given seqId have been consumed by all
-     * subscribers to the topic and no longer need to be stored. The
-     * implementation classes can decide how and if they want to garbage collect
-     * and delete these older topic messages that are no longer needed.
-     *
-     * @param topic
-     *            Topic
-     * @param seqId
-     *            Message local sequence ID
-     */
-    public void consumedUntil(ByteString topic, Long seqId);
-
-    public void setMessageBound(ByteString topic, Integer bound);
-    public void clearMessageBound(ByteString topic);
-    public void consumeToBound(ByteString topic);
-
-    /**
-     * Stop persistence manager.
-     */
-    public void stop();
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManagerWithRangeScan.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManagerWithRangeScan.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManagerWithRangeScan.java
deleted file mode 100644
index f12174f..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/PersistenceManagerWithRangeScan.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-public interface PersistenceManagerWithRangeScan extends PersistenceManager {
-    /**
-     * Executes the given range scan request
-     *
-     * @param request
-     */
-    public void scanMessages(RangeScanRequest request);
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/RangeScanRequest.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/RangeScanRequest.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/RangeScanRequest.java
deleted file mode 100644
index 3ac324d..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/RangeScanRequest.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import com.google.protobuf.ByteString;
-
-/**
- * Encapsulates a request to scan messages on the given topic starting from the
- * given seqId (included). A call-back {@link ScanCallback} is provided. As
- * messages are scanned, the relevant methods of the {@link ScanCallback} are
- * called. Two hints are provided as to when scanning should stop: in terms of
- * number of messages scanned, or in terms of the total size of messages
- * scanned. Scanning stops whenever one of these limits is exceeded. These
- * checks, especially the one about message size, are only approximate. The
- * {@link ScanCallback} used should be prepared to deal with more or less
- * messages scanned. If an error occurs during scanning, the
- * {@link ScanCallback} is notified of the error.
- *
- */
-public class RangeScanRequest {
-    ByteString topic;
-    long startSeqId;
-    int messageLimit;
-    long sizeLimit;
-    ScanCallback callback;
-    Object ctx;
-
-    public RangeScanRequest(ByteString topic, long startSeqId, int messageLimit, long sizeLimit, ScanCallback callback,
-                            Object ctx) {
-        this.topic = topic;
-        this.startSeqId = startSeqId;
-        this.messageLimit = messageLimit;
-        this.sizeLimit = sizeLimit;
-        this.callback = callback;
-        this.ctx = ctx;
-    }
-
-    public ByteString getTopic() {
-        return topic;
-    }
-
-    public long getStartSeqId() {
-        return startSeqId;
-    }
-
-    public int getMessageLimit() {
-        return messageLimit;
-    }
-
-    public long getSizeLimit() {
-        return sizeLimit;
-    }
-
-    public ScanCallback getCallback() {
-        return callback;
-    }
-
-    public Object getCtx() {
-        return ctx;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCache.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCache.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCache.java
deleted file mode 100644
index 48be3e8..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCache.java
+++ /dev/null
@@ -1,865 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.SortedSet;
-import java.util.TreeMap;
-import java.util.TreeSet;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.bookkeeper.util.MathUtils;
-import org.apache.bookkeeper.util.OrderedSafeExecutor;
-import org.apache.bookkeeper.util.SafeRunnable;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.exceptions.PubSubException.ServerNotResponsibleForTopicException;
-import org.apache.hedwig.protocol.PubSubProtocol;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
-import org.apache.hedwig.protoextensions.MessageIdUtils;
-import org.apache.hedwig.server.common.ServerConfiguration;
-import org.apache.hedwig.server.jmx.HedwigJMXService;
-import org.apache.hedwig.server.jmx.HedwigMBeanInfo;
-import org.apache.hedwig.server.jmx.HedwigMBeanRegistry;
-import org.apache.hedwig.util.Callback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.protobuf.ByteString;
-
-public class ReadAheadCache implements PersistenceManager, HedwigJMXService {
-
-    private static final Logger logger = LoggerFactory.getLogger(ReadAheadCache.class);
-
-    protected interface CacheRequest {
-        public void performRequest();
-    }
-
-    /**
-     * The underlying persistence manager that will be used for persistence and
-     * scanning below the cache
-     */
-    protected PersistenceManagerWithRangeScan realPersistenceManager;
-
-    /**
-     * The structure for the cache
-     */
-    protected ConcurrentMap<CacheKey, CacheValue> cache =
-        new ConcurrentHashMap<CacheKey, CacheValue>();
-
-    /**
-     * We also want to track the entries in seq-id order so that we can clean up
-     * entries after the last subscriber
-     */
-    protected ConcurrentMap<ByteString, SortedSet<Long>> orderedIndexOnSeqId =
-        new ConcurrentHashMap<ByteString, SortedSet<Long>>();
-
-    /**
-     * Partition Cache into Serveral Segments for simplify synchronization.
-     * Each segment maintains its time index and segment size.
-     */
-    static class CacheSegment {
-
-        /**
-         * We want to keep track of when entries were added in the cache, so that we
-         * can remove them in a FIFO fashion
-         */
-        protected SortedMap<Long, Set<CacheKey>> timeIndexOfAddition = new TreeMap<Long, Set<CacheKey>>();
-
-        /**
-         * We maintain an estimate of the current size of each cache segment,
-         * so that the thread know when to evict entries from cache segment.
-         */
-        protected AtomicLong presentSegmentSize = new AtomicLong(0);
-
-    }
-
-    /**
-     * We maintain an estimate of the current size of the cache, so that we know
-     * when to evict entries.
-     */
-    protected AtomicLong presentCacheSize = new AtomicLong(0);
-
-    /**
-     * Num pending requests.
-     */
-    protected AtomicInteger numPendingRequests = new AtomicInteger(0);
-
-    /**
-     * Cache segment for different threads
-     */
-    protected final ThreadLocal<CacheSegment> cacheSegment =
-        new ThreadLocal<CacheSegment>() {
-            @Override
-            protected CacheSegment initialValue() {
-                return new CacheSegment();
-            }
-        };
-
-    /**
-     * One instance of a callback that we will pass to the underlying
-     * persistence manager when asking it to persist messages
-     */
-    protected PersistCallback persistCallbackInstance = new PersistCallback();
-
-    /**
-     * 2 kinds of exceptions that we will use to signal error from readahead
-     */
-    protected NoSuchSeqIdException noSuchSeqIdExceptionInstance = new NoSuchSeqIdException();
-    protected ReadAheadException readAheadExceptionInstance = new ReadAheadException();
-
-    protected ServerConfiguration cfg;
-    // Boolean indicating if this thread should continue running. This is used
-    // when we want to stop the thread during a PubSubServer shutdown.
-    protected volatile boolean keepRunning = true;
-
-    protected final OrderedSafeExecutor cacheWorkers;
-    protected final int numCacheWorkers;
-    protected volatile long maxSegmentSize;
-    protected volatile long cacheEntryTTL;
-
-    // JMX Beans
-    ReadAheadCacheBean jmxCacheBean = null;
-
-    /**
-     * Constructor. Starts the cache maintainer thread
-     *
-     * @param realPersistenceManager
-     */
-    public ReadAheadCache(PersistenceManagerWithRangeScan realPersistenceManager, ServerConfiguration cfg) {
-        this.realPersistenceManager = realPersistenceManager;
-        this.cfg = cfg;
-        numCacheWorkers = cfg.getNumReadAheadCacheThreads();
-        cacheWorkers = OrderedSafeExecutor.newBuilder()
-                .name("ReadAheadCacheScheduler")
-                .numThreads(numCacheWorkers)
-                .build();
-        reloadConf(cfg);
-    }
-
-    /**
-     * Reload configuration
-     *
-     * @param conf
-     *          Server configuration object
-     */
-    protected void reloadConf(ServerConfiguration cfg) {
-        maxSegmentSize = cfg.getMaximumCacheSize() / numCacheWorkers;
-        cacheEntryTTL = cfg.getCacheEntryTTL();
-    }
-
-    public ReadAheadCache start() {
-        return this;
-    }
-
-    /**
-     * ========================================================================
-     * Methods of {@link PersistenceManager} that we will pass straight down to
-     * the real persistence manager.
-     */
-
-    @Override
-    public long getSeqIdAfterSkipping(ByteString topic, long seqId, int skipAmount) {
-        return realPersistenceManager.getSeqIdAfterSkipping(topic, seqId, skipAmount);
-    }
-
-    @Override
-    public MessageSeqId getCurrentSeqIdForTopic(ByteString topic) throws ServerNotResponsibleForTopicException {
-        return realPersistenceManager.getCurrentSeqIdForTopic(topic);
-    }
-
-    /**
-     * ========================================================================
-     * Other methods of {@link PersistenceManager} that the cache needs to take
-     * some action on.
-     *
-     * 1. Persist: We pass it through to the real persistence manager but insert
-     * our callback on the return path
-     *
-     */
-    @Override
-    public void persistMessage(PersistRequest request) {
-        // make a new PersistRequest object so that we can insert our own
-        // callback in the middle. Assign the original request as the context
-        // for the callback.
-
-        PersistRequest newRequest = new PersistRequest(request.getTopic(), request.getMessage(),
-                persistCallbackInstance, request);
-        realPersistenceManager.persistMessage(newRequest);
-    }
-
-    /**
-     * The callback that we insert on the persist request return path. The
-     * callback simply forms a {@link PersistResponse} object and inserts it in
-     * the request queue to be handled serially by the cache maintainer thread.
-     *
-     */
-    public class PersistCallback implements Callback<PubSubProtocol.MessageSeqId> {
-
-        /**
-         * In case there is a failure in persisting, just pass it to the
-         * original callback
-         */
-        @Override
-        public void operationFailed(Object ctx, PubSubException exception) {
-            PersistRequest originalRequest = (PersistRequest) ctx;
-            Callback<PubSubProtocol.MessageSeqId> originalCallback = originalRequest.getCallback();
-            Object originalContext = originalRequest.getCtx();
-            originalCallback.operationFailed(originalContext, exception);
-        }
-
-        /**
-         * When the persist finishes, we first notify the original callback of
-         * success, and then opportunistically treat the message as if it just
-         * came in through a scan
-         */
-        @Override
-        public void operationFinished(Object ctx, PubSubProtocol.MessageSeqId resultOfOperation) {
-            PersistRequest originalRequest = (PersistRequest) ctx;
-
-            // Lets call the original callback first so that the publisher can
-            // hear success
-            originalRequest.getCallback().operationFinished(originalRequest.getCtx(), resultOfOperation);
-
-            // Original message that was persisted didn't have the local seq-id.
-            // Lets add that in
-            Message messageWithLocalSeqId = MessageIdUtils.mergeLocalSeqId(originalRequest.getMessage(),
-                                            resultOfOperation.getLocalComponent());
-
-            // Now enqueue a request to add this newly persisted message to our
-            // cache
-            CacheKey cacheKey = new CacheKey(originalRequest.getTopic(), resultOfOperation.getLocalComponent());
-
-            enqueueWithoutFailureByTopic(cacheKey.getTopic(),
-                    new ScanResponse(cacheKey, messageWithLocalSeqId));
-        }
-
-    }
-
-    protected void enqueueWithoutFailureByTopic(ByteString topic, final CacheRequest obj) {
-        if (!keepRunning) {
-            return;
-        }
-        try {
-            numPendingRequests.incrementAndGet();
-            cacheWorkers.submitOrdered(topic, new SafeRunnable() {
-                @Override
-                public void safeRun() {
-                    numPendingRequests.decrementAndGet();
-                    obj.performRequest();
-                }
-            });
-        } catch (RejectedExecutionException ree) {
-            logger.error("Failed to submit cache request for topic " + topic.toStringUtf8() + " : ", ree);
-        }
-    }
-
-    /**
-     * Another method from {@link PersistenceManager}.
-     *
-     * 2. Scan - Since the scan needs to touch the cache, we will just enqueue
-     * the scan request and let the cache maintainer thread handle it.
-     */
-    @Override
-    public void scanSingleMessage(ScanRequest request) {
-        // Let the scan requests be serialized through the queue
-        enqueueWithoutFailureByTopic(request.getTopic(),
-                new ScanRequestWrapper(request));
-    }
-
-    /**
-     * Another method from {@link PersistenceManager}.
-     *
-     * 3. Enqueue the request so that the cache maintainer thread can delete all
-     * message-ids older than the one specified
-     */
-    @Override
-    public void deliveredUntil(ByteString topic, Long seqId) {
-        enqueueWithoutFailureByTopic(topic, new DeliveredUntil(topic, seqId));
-    }
-
-    /**
-     * Another method from {@link PersistenceManager}.
-     *
-     * Since this is a cache layer on top of an underlying persistence manager,
-     * we can just call the consumedUntil method there. The messages older than
-     * the latest one passed here won't be accessed anymore so they should just
-     * get aged out of the cache eventually. For now, there is no need to
-     * proactively remove those entries from the cache.
-     */
-    @Override
-    public void consumedUntil(ByteString topic, Long seqId) {
-        realPersistenceManager.consumedUntil(topic, seqId);
-    }
-
-    @Override
-    public void setMessageBound(ByteString topic, Integer bound) {
-        realPersistenceManager.setMessageBound(topic, bound);
-    }
-
-    @Override
-    public void clearMessageBound(ByteString topic) {
-        realPersistenceManager.clearMessageBound(topic);
-    }
-
-    @Override
-    public void consumeToBound(ByteString topic) {
-        realPersistenceManager.consumeToBound(topic);
-    }
-
-    /**
-     * Stop the readahead cache.
-     */
-    @Override
-    public void stop() {
-        try {
-            keepRunning = false;
-            cacheWorkers.shutdown();
-        } catch (Exception e) {
-            logger.warn("Failed to shut down cache workers : ", e);
-        }
-    }
-
-    /**
-     * The readahead policy is simple: We check if an entry already exists for
-     * the message being requested. If an entry exists, it means that either
-     * that message is already in the cache, or a read for that message is
-     * outstanding. In that case, we look a little ahead (by readAheadCount/2)
-     * and issue a range read of readAheadCount/2 messages. The idea is to
-     * ensure that the next readAheadCount messages are always available.
-     *
-     * @return the range scan that should be issued for read ahead
-     */
-    protected RangeScanRequest doReadAhead(ScanRequest request) {
-        ByteString topic = request.getTopic();
-        Long seqId = request.getStartSeqId();
-
-        int readAheadCount = cfg.getReadAheadCount();
-        // To prevent us from getting screwed by bad configuration
-        readAheadCount = Math.max(1, readAheadCount);
-
-        RangeScanRequest readAheadRequest = doReadAheadStartingFrom(topic, seqId, readAheadCount);
-
-        if (readAheadRequest != null) {
-            return readAheadRequest;
-        }
-
-        // start key was already there in the cache so no readahead happened,
-        // lets look a little beyond
-        seqId = realPersistenceManager.getSeqIdAfterSkipping(topic, seqId, readAheadCount / 2);
-
-        readAheadRequest = doReadAheadStartingFrom(topic, seqId, readAheadCount / 2);
-
-        return readAheadRequest;
-    }
-
-    /**
-     * This method just checks if the provided seq-id already exists in the
-     * cache. If not, a range read of the specified amount is issued.
-     *
-     * @param topic
-     * @param seqId
-     * @param readAheadCount
-     * @return The range read that should be issued
-     */
-    protected RangeScanRequest doReadAheadStartingFrom(ByteString topic, long seqId, int readAheadCount) {
-
-        long startSeqId = seqId;
-        Queue<CacheKey> installedStubs = new LinkedList<CacheKey>();
-
-        int i = 0;
-
-        for (; i < readAheadCount; i++) {
-            CacheKey cacheKey = new CacheKey(topic, seqId);
-
-            // Even if a stub exists, it means that a scan for that is
-            // outstanding
-            if (cache.containsKey(cacheKey)) {
-                break;
-            }
-            CacheValue cacheValue = new CacheValue();
-            if (null != cache.putIfAbsent(cacheKey, cacheValue)) {
-                logger.warn("It is unexpected that more than one threads are adding message to cache key {}"
-                            +" at the same time.", cacheKey);
-            }
-
-            logger.debug("Adding cache stub for: {}", cacheKey);
-            installedStubs.add(cacheKey);
-
-            seqId = realPersistenceManager.getSeqIdAfterSkipping(topic, seqId, 1);
-        }
-
-        // so how many did we decide to readahead
-        if (i == 0) {
-            // no readahead, hence return false
-            return null;
-        }
-
-        long readAheadSizeLimit = cfg.getReadAheadSizeBytes();
-        ReadAheadScanCallback callback = new ReadAheadScanCallback(installedStubs, topic);
-        RangeScanRequest rangeScanRequest = new RangeScanRequest(topic, startSeqId, i, readAheadSizeLimit, callback,
-                null);
-
-        return rangeScanRequest;
-
-    }
-
-    /**
-     * This is the callback that is used for the range scans.
-     */
-    protected class ReadAheadScanCallback implements ScanCallback {
-        Queue<CacheKey> installedStubs;
-        ByteString topic;
-
-        /**
-         * Constructor
-         *
-         * @param installedStubs
-         *            The list of stubs that were installed for this range scan
-         * @param topic
-         */
-        public ReadAheadScanCallback(Queue<CacheKey> installedStubs, ByteString topic) {
-            this.installedStubs = installedStubs;
-            this.topic = topic;
-        }
-
-        @Override
-        public void messageScanned(Object ctx, Message message) {
-
-            // Any message we read is potentially useful for us, so lets first
-            // enqueue it
-            CacheKey cacheKey = new CacheKey(topic, message.getMsgId().getLocalComponent());
-            enqueueWithoutFailureByTopic(topic, new ScanResponse(cacheKey, message));
-
-            // Now lets see if this message is the one we were expecting
-            CacheKey expectedKey = installedStubs.peek();
-
-            if (expectedKey == null) {
-                // Was not expecting any more messages to come in, but they came
-                // in so we will keep them
-                return;
-            }
-
-            if (expectedKey.equals(cacheKey)) {
-                // what we got is what we expected, dequeue it so we get the
-                // next expected one
-                installedStubs.poll();
-                return;
-            }
-
-            // If reached here, what we scanned was not what we were expecting.
-            // This means that we have wrong stubs installed in the cache. We
-            // should remove them, so that whoever is waiting on them can retry.
-            // This shouldn't be happening usually
-            logger.warn("Unexpected message seq-id: " + message.getMsgId().getLocalComponent() + " on topic: "
-                        + topic.toStringUtf8() + " from readahead scan, was expecting seq-id: " + expectedKey.seqId
-                        + " topic: " + expectedKey.topic.toStringUtf8() + " installedStubs: " + installedStubs);
-            enqueueDeleteOfRemainingStubs(noSuchSeqIdExceptionInstance);
-
-        }
-
-        @Override
-        public void scanFailed(Object ctx, Exception exception) {
-            enqueueDeleteOfRemainingStubs(exception);
-        }
-
-        @Override
-        public void scanFinished(Object ctx, ReasonForFinish reason) {
-            // If the scan finished because no more messages are present, its ok
-            // to leave the stubs in place because they will get filled in as
-            // new publishes happen. However, if the scan finished due to some
-            // other reason, e.g., read ahead size limit was reached, we want to
-            // delete the stubs, so that when the time comes, we can schedule
-            // another readahead request.
-            if (reason != ReasonForFinish.NO_MORE_MESSAGES) {
-                enqueueDeleteOfRemainingStubs(readAheadExceptionInstance);
-            }
-        }
-
-        private void enqueueDeleteOfRemainingStubs(Exception reason) {
-            CacheKey installedStub;
-            while ((installedStub = installedStubs.poll()) != null) {
-                enqueueWithoutFailureByTopic(installedStub.getTopic(),
-                        new ExceptionOnCacheKey(installedStub, reason));
-            }
-        }
-    }
-
-    protected static class HashSetCacheKeyFactory implements Factory<Set<CacheKey>> {
-        protected final static HashSetCacheKeyFactory instance = new HashSetCacheKeyFactory();
-
-        @Override
-        public Set<CacheKey> newInstance() {
-            return new HashSet<CacheKey>();
-        }
-    }
-
-    protected static class TreeSetLongFactory implements Factory<SortedSet<Long>> {
-        protected final static TreeSetLongFactory instance = new TreeSetLongFactory();
-
-        @Override
-        public SortedSet<Long> newInstance() {
-            return new TreeSet<Long>();
-        }
-    }
-
-    /**
-     * For adding the message to the cache, we do some bookeeping such as the
-     * total size of cache, order in which entries were added etc. If the size
-     * of the cache has exceeded our budget, old entries are collected.
-     *
-     * @param cacheKey
-     * @param message
-     */
-    protected void addMessageToCache(final CacheKey cacheKey,
-                                     final Message message, final long currTime) {
-        logger.debug("Adding msg {} to readahead cache", cacheKey);
-
-        CacheValue cacheValue;
-        if ((cacheValue = cache.get(cacheKey)) == null) {
-            cacheValue = new CacheValue();
-            CacheValue oldValue = cache.putIfAbsent(cacheKey, cacheValue);
-            if (null != oldValue) {
-                logger.warn("Weird! Should not have two threads adding message to cache key {} at the same time.",
-                            cacheKey);
-                cacheValue = oldValue;
-            }
-        }
-
-        CacheSegment segment = cacheSegment.get();
-        if (cacheValue.isStub()) { // update cache size only when cache value is a stub
-            int size = message.getBody().size();
-
-            // update the cache size
-            segment.presentSegmentSize.addAndGet(size);
-            presentCacheSize.addAndGet(size);
-        }
-
-        synchronized (cacheValue) {
-            // finally add the message to the cache
-            cacheValue.setMessageAndInvokeCallbacks(message, currTime);
-        }
-
-        // maintain the index of seq-id
-        // no lock since threads are partitioned by topics
-        MapMethods.addToMultiMap(orderedIndexOnSeqId, cacheKey.getTopic(),
-                                 cacheKey.getSeqId(), TreeSetLongFactory.instance);
-
-        // maintain the time index of addition
-        MapMethods.addToMultiMap(segment.timeIndexOfAddition, currTime,
-                                 cacheKey, HashSetCacheKeyFactory.instance);
-
-        collectOldOrExpiredCacheEntries(segment);
-    }
-
-    protected void removeMessageFromCache(final CacheKey cacheKey, Exception exception,
-                                          final boolean maintainTimeIndex,
-                                          final boolean maintainSeqIdIndex) {
-        CacheValue cacheValue = cache.remove(cacheKey);
-
-        if (cacheValue == null) {
-            return;
-        }
-
-        CacheSegment segment = cacheSegment.get();
-
-        long timeOfAddition = 0;
-        synchronized (cacheValue) {
-            if (cacheValue.isStub()) {
-                cacheValue.setErrorAndInvokeCallbacks(exception);
-                // Stubs are not present in the indexes, so don't need to maintain
-                // indexes here
-                return;
-            }
-
-            int size = 0 - cacheValue.getMessage().getBody().size();
-            presentCacheSize.addAndGet(size);
-            segment.presentSegmentSize.addAndGet(size);
-            timeOfAddition = cacheValue.getTimeOfAddition();
-        }
-
-        if (maintainSeqIdIndex) {
-            MapMethods.removeFromMultiMap(orderedIndexOnSeqId, cacheKey.getTopic(),
-                                          cacheKey.getSeqId());
-        }
-        if (maintainTimeIndex) {
-            MapMethods.removeFromMultiMap(segment.timeIndexOfAddition,
-                                          timeOfAddition, cacheKey);
-        }
-    }
-
-    /**
-     * Collection of old entries is simple. Just collect in insert-time order,
-     * oldest to newest.
-     */
-    protected void collectOldOrExpiredCacheEntries(CacheSegment segment) {
-        if (cacheEntryTTL > 0) {
-            // clear expired entries
-            while (!segment.timeIndexOfAddition.isEmpty()) {
-                Long earliestTime = segment.timeIndexOfAddition.firstKey();
-                if (MathUtils.now() - earliestTime < cacheEntryTTL) {
-                    break;
-                }
-                collectCacheEntriesAtTimestamp(segment, earliestTime);
-            }
-        }
-
-        while (segment.presentSegmentSize.get() > maxSegmentSize &&
-               !segment.timeIndexOfAddition.isEmpty()) {
-            Long earliestTime = segment.timeIndexOfAddition.firstKey();
-            collectCacheEntriesAtTimestamp(segment, earliestTime);
-        }
-    }
-
-    private void collectCacheEntriesAtTimestamp(CacheSegment segment, long timestamp) {
-        Set<CacheKey> oldCacheEntries = segment.timeIndexOfAddition.get(timestamp);
-
-        // Note: only concrete cache entries, and not stubs are in the time
-        // index. Hence there can be no callbacks pending on these cache
-        // entries. Hence safe to remove them directly.
-        for (Iterator<CacheKey> iter = oldCacheEntries.iterator(); iter.hasNext();) {
-            final CacheKey cacheKey = iter.next();
-
-            logger.debug("Removing {} from cache because it's the oldest.", cacheKey);
-            removeMessageFromCache(cacheKey, readAheadExceptionInstance, //
-                                   // maintainTimeIndex=
-                                   false,
-                                   // maintainSeqIdIndex=
-                                   true);
-        }
-
-        segment.timeIndexOfAddition.remove(timestamp);
-    }
-
-    /**
-     * ========================================================================
-     * The rest is just simple wrapper classes.
-     *
-     */
-
-    protected class ExceptionOnCacheKey implements CacheRequest {
-        CacheKey cacheKey;
-        Exception exception;
-
-        public ExceptionOnCacheKey(CacheKey cacheKey, Exception exception) {
-            this.cacheKey = cacheKey;
-            this.exception = exception;
-        }
-
-        /**
-         * If for some reason, an outstanding read on a cache stub fails,
-         * exception for that key is enqueued by the
-         * {@link ReadAheadScanCallback}. To handle this, we simply send error
-         * on the callbacks registered for that stub, and delete the entry from
-         * the cache
-         */
-        @Override
-        public void performRequest() {
-            removeMessageFromCache(cacheKey, exception,
-                                   // maintainTimeIndex=
-                                   true,
-                                   // maintainSeqIdIndex=
-                                   true);
-        }
-
-    }
-
-    @SuppressWarnings("serial")
-    protected static class NoSuchSeqIdException extends Exception {
-
-        public NoSuchSeqIdException() {
-            super("No such seq-id");
-        }
-    }
-
-    @SuppressWarnings("serial")
-    protected static class ReadAheadException extends Exception {
-        public ReadAheadException() {
-            super("Readahead failed");
-        }
-    }
-
-    public class CancelScanRequestOp implements CacheRequest {
-
-        final CancelScanRequest request;
-
-        public CancelScanRequestOp(CancelScanRequest request) {
-            this.request = request;
-        }
-
-        @Override
-        public void performRequest() {
-            // cancel scan request
-            cancelScanRequest(request.getScanRequest());
-        }
-
-        void cancelScanRequest(ScanRequest request) {
-            if (null == request) {
-                // nothing to cancel
-                return;
-            }
-
-            CacheKey cacheKey = new CacheKey(request.getTopic(), request.getStartSeqId());
-            CacheValue cacheValue = cache.get(cacheKey);
-            if (null == cacheValue) {
-                // cache value is evicted
-                // so it's callback would be called, we don't need to worry about
-                // cancel it. since it was treated as executed.
-                return;
-            }
-            cacheValue.removeCallback(request.getCallback(), request.getCtx());
-        }
-    }
-
-    public void cancelScanRequest(ByteString topic, CancelScanRequest request) {
-        enqueueWithoutFailureByTopic(topic, new CancelScanRequestOp(request));
-    }
-
-    protected class ScanResponse implements CacheRequest {
-        CacheKey cacheKey;
-        Message message;
-
-        public ScanResponse(CacheKey cacheKey, Message message) {
-            this.cacheKey = cacheKey;
-            this.message = message;
-        }
-
-        @Override
-        public void performRequest() {
-            addMessageToCache(cacheKey, message, MathUtils.now());
-        }
-
-    }
-
-    protected class DeliveredUntil implements CacheRequest {
-        ByteString topic;
-        Long seqId;
-
-        public DeliveredUntil(ByteString topic, Long seqId) {
-            this.topic = topic;
-            this.seqId = seqId;
-        }
-
-        @Override
-        public void performRequest() {
-            SortedSet<Long> orderedSeqIds = orderedIndexOnSeqId.get(topic);
-            if (orderedSeqIds == null) {
-                return;
-            }
-
-            // focus on the set of messages with seq-ids <= the one that
-            // has been delivered until
-            SortedSet<Long> headSet = orderedSeqIds.headSet(seqId + 1);
-
-            for (Iterator<Long> iter = headSet.iterator(); iter.hasNext();) {
-                Long seqId = iter.next();
-                CacheKey cacheKey = new CacheKey(topic, seqId);
-
-                logger.debug("Removing {} from cache because every subscriber has moved past",
-                    cacheKey);
-
-                removeMessageFromCache(cacheKey, readAheadExceptionInstance, //
-                                       // maintainTimeIndex=
-                                       true,
-                                       // maintainSeqIdIndex=
-                                       false);
-                iter.remove();
-            }
-
-            if (orderedSeqIds.isEmpty()) {
-                orderedIndexOnSeqId.remove(topic);
-            }
-        }
-    }
-
-    protected class ScanRequestWrapper implements CacheRequest {
-        ScanRequest request;
-
-        public ScanRequestWrapper(ScanRequest request) {
-            this.request = request;
-        }
-
-        /**
-         * To handle a scan request, we first try to do readahead (which might
-         * cause a range read to be issued to the underlying persistence
-         * manager). The readahead will put a stub in the cache, if the message
-         * is not already present in the cache. The scan callback that is part
-         * of the scan request is added to this stub, and will be called later
-         * when the message arrives as a result of the range scan issued to the
-         * underlying persistence manager.
-         */
-
-        @Override
-        public void performRequest() {
-
-            RangeScanRequest readAheadRequest = doReadAhead(request);
-
-            // Read ahead must have installed at least a stub for us, so this
-            // can't be null
-            CacheKey cacheKey = new CacheKey(request.getTopic(), request.getStartSeqId());
-            CacheValue cacheValue = cache.get(cacheKey);
-            if (null == cacheValue) {
-                logger.error("Cache key {} is removed after installing stub when scanning.", cacheKey);
-                // reissue the request
-                scanSingleMessage(request);
-                return;
-            }
-
-            synchronized (cacheValue) {
-                // Add our callback to the stub. If the cache value was already a
-                // concrete message, the callback will be called right away
-                cacheValue.addCallback(request.getCallback(), request.getCtx());
-            }
-
-            if (readAheadRequest != null) {
-                realPersistenceManager.scanMessages(readAheadRequest);
-            }
-        }
-    }
-
-    @Override
-    public void registerJMX(HedwigMBeanInfo parent) {
-        try {
-            jmxCacheBean = new ReadAheadCacheBean(this);
-            HedwigMBeanRegistry.getInstance().register(jmxCacheBean, parent);
-        } catch (Exception e) {
-            logger.warn("Failed to register readahead cache with JMX", e);
-            jmxCacheBean = null;
-        }
-    }
-
-    @Override
-    public void unregisterJMX() {
-        try {
-            if (jmxCacheBean != null) {
-                HedwigMBeanRegistry.getInstance().unregister(jmxCacheBean);
-            }
-        } catch (Exception e) {
-            logger.warn("Failed to unregister readahead cache with JMX", e);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheBean.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheBean.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheBean.java
deleted file mode 100644
index 1f43095..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheBean.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import org.apache.hedwig.server.jmx.HedwigMBeanInfo;
-
-/**
- * Read Ahead Cache Bean
- */
-public class ReadAheadCacheBean implements ReadAheadCacheMXBean,
-        HedwigMBeanInfo {
-
-    ReadAheadCache cache;
-    public ReadAheadCacheBean(ReadAheadCache cache) {
-        this.cache = cache;
-    }
-
-    @Override
-    public String getName() {
-        return "ReadAheadCache";
-    }
-
-    @Override
-    public boolean isHidden() {
-        return false;
-    }
-
-    @Override
-    public long getMaxCacheSize() {
-        return cache.cfg.getMaximumCacheSize();
-    }
-
-    @Override
-    public long getPresentCacheSize() {
-        return cache.presentCacheSize.get();
-    }
-
-    @Override
-    public int getNumCachedEntries() {
-        return cache.cache.size();
-    }
-
-    @Override
-    public int getNumPendingCacheRequests() {
-        return cache.numPendingRequests.get();
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheMXBean.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheMXBean.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheMXBean.java
deleted file mode 100644
index eba77a0..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ReadAheadCacheMXBean.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-/**
- * Read Ahead Cache MBean
- */
-public interface ReadAheadCacheMXBean {
-
-    /**
-     * @return max cache size
-     */
-    public long getMaxCacheSize();
-
-    /**
-     * @return present cache size
-     */
-    public long getPresentCacheSize();
-
-    /**
-     * @return number of cached entries
-     */
-    public int getNumCachedEntries();
-
-    /**
-     * @return number of pending cache requests
-     */
-    public int getNumPendingCacheRequests();
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallback.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallback.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallback.java
deleted file mode 100644
index 42ebb93..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallback.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-
-public interface ScanCallback {
-
-    enum ReasonForFinish {
-        NO_MORE_MESSAGES, SIZE_LIMIT_EXCEEDED, NUM_MESSAGES_LIMIT_EXCEEDED
-    };
-
-    /**
-     * This method is called when a message is read from the persistence layer
-     * as part of a scan. The message just read is handed to this listener which
-     * can then take the desired action on it. The return value from the method
-     * indicates whether the scan should continue or not.
-     *
-     * @param ctx
-     *            The context for the callback
-     * @param message
-     *            The message just scanned from the log
-     * @return true if the scan should continue, false otherwise
-     */
-    public void messageScanned(Object ctx, Message message);
-
-    /**
-     * This method is called when the scan finishes
-     *
-     *
-     * @param ctx
-     * @param reason
-     */
-
-    public abstract void scanFinished(Object ctx, ReasonForFinish reason);
-
-    /**
-     * This method is called when the operation failed due to some reason. The
-     * reason for failure is passed in.
-     *
-     * @param ctx
-     *            The context for the callback
-     * @param exception
-     *            The reason for the failure of the scan
-     */
-    public abstract void scanFailed(Object ctx, Exception exception);
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallbackWithContext.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallbackWithContext.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallbackWithContext.java
deleted file mode 100644
index a39a197..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanCallbackWithContext.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-public class ScanCallbackWithContext {
-    ScanCallback scanCallback;
-    Object ctx;
-
-    public ScanCallbackWithContext(ScanCallback callback, Object ctx) {
-        this.scanCallback = callback;
-        this.ctx = ctx;
-    }
-
-    public ScanCallback getScanCallback() {
-        return scanCallback;
-    }
-
-    public Object getCtx() {
-        return ctx;
-    }
-
-    @Override
-    public boolean equals(Object other) {
-        if (!(other instanceof ScanCallbackWithContext)) {
-            return false;
-        }
-        ScanCallbackWithContext otherCb =
-            (ScanCallbackWithContext) other;
-        // Ensure that it was same callback & same ctx
-        return scanCallback == otherCb.scanCallback &&
-               ctx == otherCb.ctx;
-    }
-
-    @Override
-    public int hashCode() {
-        return scanCallback.hashCode();
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanRequest.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanRequest.java b/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanRequest.java
deleted file mode 100644
index c985840..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/persistence/ScanRequest.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- * 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.hedwig.server.persistence;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.protocol.PubSubProtocol.Message;
-
-/**
- * Encapsulates a request for reading a single message. The message on the given
- * topic <b>at</b> the given seqId is scanned. A call-back {@link ScanCallback}
- * is provided. When the message is scanned, the
- * {@link ScanCallback#messageScanned(Object, Message)} method is called. Since
- * there is only 1 record to be scanned the
- * {@link ScanCallback#operationFinished(Object)} method may not be called since
- * its redundant.
- * {@link ScanCallback#scanFailed(Object, org.apache.hedwig.exceptions.PubSubException)}
- * method is called in case of error.
- *
- */
-public class ScanRequest {
-    ByteString topic;
-    long startSeqId;
-    ScanCallback callback;
-    Object ctx;
-
-    public ScanRequest(ByteString topic, long startSeqId, ScanCallback callback, Object ctx) {
-        this.topic = topic;
-        this.startSeqId = startSeqId;
-        this.callback = callback;
-        this.ctx = ctx;
-    }
-
-    public ByteString getTopic() {
-        return topic;
-    }
-
-    public long getStartSeqId() {
-        return startSeqId;
-    }
-
-    public ScanCallback getCallback() {
-        return callback;
-    }
-
-    public Object getCtx() {
-        return ctx;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-server/src/main/java/org/apache/hedwig/server/proxy/ChannelTracker.java
----------------------------------------------------------------------
diff --git a/hedwig-server/src/main/java/org/apache/hedwig/server/proxy/ChannelTracker.java b/hedwig-server/src/main/java/org/apache/hedwig/server/proxy/ChannelTracker.java
deleted file mode 100644
index 10f0889..0000000
--- a/hedwig-server/src/main/java/org/apache/hedwig/server/proxy/ChannelTracker.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/**
- * 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.hedwig.server.proxy;
-
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-
-import org.jboss.netty.channel.Channel;
-
-import com.google.protobuf.ByteString;
-import org.apache.hedwig.client.api.Subscriber;
-import org.apache.hedwig.client.data.TopicSubscriber;
-import org.apache.hedwig.exceptions.PubSubException;
-import org.apache.hedwig.exceptions.PubSubException.TopicBusyException;
-import org.apache.hedwig.server.handlers.ChannelDisconnectListener;
-import org.apache.hedwig.util.Callback;
-
-public class ChannelTracker implements ChannelDisconnectListener {
-    HashMap<TopicSubscriber, Channel> topicSub2Channel = new HashMap<TopicSubscriber, Channel>();
-    HashMap<Channel, List<TopicSubscriber>> channel2TopicSubs = new HashMap<Channel, List<TopicSubscriber>>();
-    Subscriber subscriber;
-
-    public ChannelTracker(Subscriber subscriber) {
-        this.subscriber = subscriber;
-    }
-
-    static Callback<Void> noOpCallback = new Callback<Void>() {
-        public void operationFailed(Object ctx, PubSubException exception) {
-        };
-
-        public void operationFinished(Object ctx, Void resultOfOperation) {
-        };
-    };
-
-    public synchronized void channelDisconnected(Channel channel) {
-        List<TopicSubscriber> topicSubs = channel2TopicSubs.remove(channel);
-
-        if (topicSubs == null) {
-            return;
-        }
-
-        for (TopicSubscriber topicSub : topicSubs) {
-            topicSub2Channel.remove(topicSub);
-            subscriber.asyncCloseSubscription(topicSub.getTopic(), topicSub.getSubscriberId(), noOpCallback, null);
-        }
-    }
-
-    public synchronized void subscribeSucceeded(TopicSubscriber topicSubscriber, Channel channel)
-            throws TopicBusyException {
-
-        if (!channel.isConnected()) {
-            // channel got disconnected while we were processing the
-            // subscribe request, nothing much we can do in this case
-            return;
-        }
-
-        if (topicSub2Channel.containsKey(topicSubscriber)) {
-            TopicBusyException pse = new PubSubException.TopicBusyException(
-                "subscription for this topic, subscriberId is already being served on a different channel");
-            throw pse;
-        }
-
-        topicSub2Channel.put(topicSubscriber, channel);
-
-        List<TopicSubscriber> topicSubs = channel2TopicSubs.get(channel);
-
-        if (topicSubs == null) {
-            topicSubs = new LinkedList<TopicSubscriber>();
-            channel2TopicSubs.put(channel, topicSubs);
-        }
-        topicSubs.add(topicSubscriber);
-
-    }
-
-    public void aboutToCloseSubscription(ByteString topic, ByteString subscriberId) {
-        removeSubscriber(topic, subscriberId);
-    } 
-
-    public void aboutToUnsubscribe(ByteString topic, ByteString subscriberId) {
-        removeSubscriber(topic, subscriberId);
-    }
-
-    private synchronized void removeSubscriber(ByteString topic, ByteString subscriberId) {
-        TopicSubscriber topicSub = new TopicSubscriber(topic, subscriberId);
-
-        Channel channel = topicSub2Channel.remove(topicSub);
-
-        if (channel != null) {
-            List<TopicSubscriber> topicSubs = channel2TopicSubs.get(channel);
-            if (topicSubs != null) {
-                topicSubs.remove(topicSub);
-            }
-        }
-    }
-
-    public synchronized void checkChannelMatches(ByteString topic, ByteString subscriberId, Channel channel)
-            throws PubSubException {
-        Channel subscribedChannel = getChannel(topic, subscriberId);
-
-        if (subscribedChannel == null) {
-            throw new PubSubException.ClientNotSubscribedException(
-                "Can't start delivery since client is not subscribed");
-        }
-
-        if (subscribedChannel != channel) {
-            throw new PubSubException.TopicBusyException(
-                "Can't start delivery since client is subscribed on a different channel");
-        }
-
-    }
-
-    public synchronized Channel getChannel(ByteString topic, ByteString subscriberId) {
-        return topicSub2Channel.get(new TopicSubscriber(topic, subscriberId));
-    }
-
-}


Mime
View raw message