You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by fp...@apache.org on 2011/11/11 11:24:19 UTC
svn commit: r1200806 - in /zookeeper/bookkeeper/trunk: ./
bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/
bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ doc/
Author: fpj
Date: Fri Nov 11 10:24:19 2011
New Revision: 1200806
URL: http://svn.apache.org/viewvc?rev=1200806&view=rev
Log:
BOOKKEEPER-82: support journal rolling (Sijie Guo via fpj)
Added:
zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieJournalRollingTest.java
Modified:
zookeeper/bookkeeper/trunk/CHANGES.txt
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java
zookeeper/bookkeeper/trunk/doc/bookkeeperConfig.textile
Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1200806&r1=1200805&r2=1200806&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Fri Nov 11 10:24:19 2011
@@ -62,6 +62,8 @@ BUGFIXES:
BOOKKEEPER-50: NullPointException at LedgerDescriptor#cmpMasterKey (Sijie Guo via ivank)
+ BOOKKEEPER-82: support journal rolling (Sijie Guo via fpj)
+
hedwig-server/
BOOKKEEPER-43: NullPointException when releasing topic (Sijie Guo via breed)
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java?rev=1200806&r1=1200805&r2=1200806&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java Fri Nov 11 10:24:19 2011
@@ -35,7 +35,9 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
+import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.proto.BookieServer;
@@ -57,6 +59,11 @@ import org.apache.zookeeper.ZooDefs.Ids;
public class Bookie extends Thread {
HashMap<Long, LedgerDescriptor> ledgers = new HashMap<Long, LedgerDescriptor>();
static Logger LOG = Logger.getLogger(Bookie.class);
+ final static long MB = 1024 * 1024L;
+ // max journal file size
+ final static long MAX_JOURNAL_SIZE = Long.getLong("journal_max_size_mb", 2 * 1024) * MB;
+ // number journal files kept before marked journal
+ final static int MAX_BACKUP_JOURNALS = Integer.getInteger("journal_max_backups", 5);
final File journalDirectory;
@@ -101,8 +108,34 @@ public class Bookie extends Thread {
EntryLogger entryLogger;
LedgerCache ledgerCache;
+ /**
+ * SyncThread is a background thread which flushes ledger index pages periodically.
+ * Also it takes responsibility of garbage collecting journal files.
+ *
+ * <p>
+ * Before flushing, SyncThread first records a log marker {journalId, journalPos} in memory,
+ * which indicates entries before this log marker would be persisted to ledger files.
+ * Then sync thread begans flush ledger index pages to ledger index files, flush entry
+ * logger to ensure all entries persisted to entry loggers for future reads.
+ * </p>
+ * <p>
+ * After all data has been persisted to ledger index files and entry loggers, it is safe
+ * to persist the log marker to disk. If bookie failed after persist log mark,
+ * bookie is able to relay journal entries started from last log mark without lossing
+ * any entries.
+ * </p>
+ * <p>
+ * Those journal files whose id are less than the log id in last log mark, could be
+ * removed safely after persisting last log mark. We provide a setting to let user keeping
+ * number of old journal files which may be used for munually recovery in critical disaster.
+ * </p>
+ */
class SyncThread extends Thread {
volatile boolean running = true;
+ // flag to ensure sync thread will not be interrupted during flush
+ final AtomicBoolean flushing = new AtomicBoolean(false);
+ // make flush interval as a parameter
+ final int flushInterval = Integer.getInteger("flush_interval", 100);
public SyncThread() {
super("SyncThread");
}
@@ -111,7 +144,7 @@ public class Bookie extends Thread {
while(running) {
synchronized(this) {
try {
- wait(100);
+ wait(flushInterval);
if (!entryLogger.testAndClearSomethingWritten()) {
continue;
}
@@ -120,6 +153,17 @@ public class Bookie extends Thread {
continue;
}
}
+
+ // try to mark flushing flag to make sure it would not be interrupted
+ // by shutdown during flushing. otherwise it will receive
+ // ClosedByInterruptException which may cause index file & entry logger
+ // closed and corrupted.
+ if (!flushing.compareAndSet(false, true)) {
+ // set flushing flag failed, means flushing is true now
+ // indicates another thread wants to interrupt sync thread to exit
+ break;
+ }
+
lastLogMark.markLog();
try {
ledgerCache.flushLedger(true);
@@ -132,8 +176,48 @@ public class Bookie extends Thread {
LOG.error("Exception flushing entry logger", e);
}
lastLogMark.rollLog();
+
+ // list the journals whose has been marked
+ List<Long> logs = listJournalIds(journalDirectory, new JournalIdFilter() {
+ @Override
+ public boolean accept(long journalId) {
+ if (journalId < lastLogMark.lastMark.txnLogId) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+ });
+
+ // keep MAX_BACKUP_JOURNALS journal files before marked journal
+ if (logs.size() >= MAX_BACKUP_JOURNALS) {
+ int maxIdx = logs.size() - MAX_BACKUP_JOURNALS;
+ for (int i=0; i<maxIdx; i++) {
+ long id = logs.get(i);
+ // make sure the journal id is smaller than marked journal id
+ if (id < lastLogMark.lastMark.txnLogId) {
+ File journalFile = new File(journalDirectory, Long.toHexString(id) + ".txn");
+ journalFile.delete();
+ LOG.info("garbage collected journal " + journalFile.getName());
+ }
+ }
+ }
+
+ // clear flushing flag
+ flushing.set(false);
}
}
+
+ // shutdown sync thread
+ void shutdown() throws InterruptedException {
+ running = false;
+ if (flushing.compareAndSet(false, true)) {
+ // if setting flushing flag succeed, means syncThread is not flushing now
+ // it is safe to interrupt itself now
+ this.interrupt();
+ }
+ this.join();
+ }
}
SyncThread syncThread = new SyncThread();
@@ -143,63 +227,75 @@ public class Bookie extends Thread {
entryLogger = new EntryLogger(ledgerDirectories, this);
ledgerCache = new LedgerCache(ledgerDirectories);
lastLogMark.readLog();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Last Log Mark : " + lastLogMark);
+ }
final long markedLogId = lastLogMark.txnLogId;
- if (markedLogId > 0) {
- File logFiles[] = journalDirectory.listFiles();
- ArrayList<Long> logs = new ArrayList<Long>();
- for(File f: logFiles) {
- String name = f.getName();
- if (!name.endsWith(".txn")) {
- continue;
- }
- String idString = name.split("\\.")[0];
- long id = Long.parseLong(idString, 16);
- if (id < markedLogId) {
- continue;
+ List<Long> logs = listJournalIds(journalDirectory, new JournalIdFilter() {
+ @Override
+ public boolean accept(long journalId) {
+ if (journalId < markedLogId) {
+ return false;
}
- logs.add(id);
+ return true;
}
- Collections.sort(logs);
+ });
+ // last log mark may be missed due to no sync up before
+ // validate filtered log ids only when we have markedLogId
+ if (markedLogId > 0) {
if (logs.size() == 0 || logs.get(0) != markedLogId) {
throw new IOException("Recovery log " + markedLogId + " is missing");
}
- // TODO: When reading in the journal logs that need to be synced, we
- // should use BufferedChannels instead to minimize the amount of
- // system calls done.
- ByteBuffer lenBuff = ByteBuffer.allocate(4);
- ByteBuffer recBuff = ByteBuffer.allocate(64*1024);
- for(Long id: logs) {
- FileChannel recLog = openChannel(id);
- while(true) {
- lenBuff.clear();
- fullRead(recLog, lenBuff);
- if (lenBuff.remaining() != 0) {
- break;
- }
- lenBuff.flip();
- int len = lenBuff.getInt();
- if (len == 0) {
- break;
- }
- recBuff.clear();
- if (recBuff.remaining() < len) {
- recBuff = ByteBuffer.allocate(len);
- }
- recBuff.limit(len);
- if (fullRead(recLog, recBuff) != len) {
- // This seems scary, but it just means that this is where we
- // left off writing
- break;
- }
- recBuff.flip();
- long ledgerId = recBuff.getLong();
- LedgerDescriptor handle = getHandle(ledgerId, false);
- try {
- recBuff.rewind();
- handle.addEntry(recBuff);
- } finally {
- putHandle(handle);
- }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Try to relay journal logs : " + logs);
+ }
+ // TODO: When reading in the journal logs that need to be synced, we
+ // should use BufferedChannels instead to minimize the amount of
+ // system calls done.
+ ByteBuffer lenBuff = ByteBuffer.allocate(4);
+ ByteBuffer recBuff = ByteBuffer.allocate(64*1024);
+ for(Long id: logs) {
+ FileChannel recLog ;
+ if(id == markedLogId) {
+ long markedLogPosition = lastLogMark.txnLogPosition;
+ recLog = openChannel(id, markedLogPosition);
+ } else {
+ recLog = openChannel(id);
+ }
+
+ while(true) {
+ lenBuff.clear();
+ fullRead(recLog, lenBuff);
+ if (lenBuff.remaining() != 0) {
+ break;
+ }
+ lenBuff.flip();
+ int len = lenBuff.getInt();
+ if (len == 0) {
+ break;
+ }
+ recBuff.clear();
+ if (recBuff.remaining() < len) {
+ recBuff = ByteBuffer.allocate(len);
+ }
+ recBuff.limit(len);
+ if (fullRead(recLog, recBuff) != len) {
+ // This seems scary, but it just means that this is where we
+ // left off writing
+ break;
+ }
+ recBuff.flip();
+ long ledgerId = recBuff.getLong();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Relay journal - ledger id : " + ledgerId);
+ }
+ LedgerDescriptor handle = getHandle(ledgerId, false);
+ try {
+ recBuff.rewind();
+ handle.addEntry(recBuff);
+ } finally {
+ putHandle(handle);
}
}
}
@@ -214,6 +310,39 @@ public class Bookie extends Thread {
running = true;
}
+ public static interface JournalIdFilter {
+ public boolean accept(long journalId);
+ }
+
+ /**
+ * List all journal ids by a specified journal id filer
+ *
+ * @param journalDir journal dir
+ * @param filter journal id filter
+ * @return list of filtered ids
+ */
+ public static List<Long> listJournalIds(File journalDir, JournalIdFilter filter) {
+ File logFiles[] = journalDir.listFiles();
+ List<Long> logs = new ArrayList<Long>();
+ for(File f: logFiles) {
+ String name = f.getName();
+ if (!name.endsWith(".txn")) {
+ continue;
+ }
+ String idString = name.split("\\.")[0];
+ long id = Long.parseLong(idString, 16);
+ if (filter != null) {
+ if (filter.accept(id)) {
+ logs.add(id);
+ }
+ } else {
+ logs.add(id);
+ }
+ }
+ Collections.sort(logs);
+ return logs;
+ }
+
/**
* Instantiate the ZooKeeper client for the Bookie.
*/
@@ -426,8 +555,14 @@ public class Bookie extends Thread {
synchronized void rollLog() {
byte buff[] = new byte[16];
ByteBuffer bb = ByteBuffer.wrap(buff);
- bb.putLong(txnLogId);
- bb.putLong(txnLogPosition);
+ // we should record <logId, logPosition> marked in markLog
+ // which is safe since records before lastMark have been
+ // persisted to disk (both index & entry logger)
+ bb.putLong(lastMark.txnLogId);
+ bb.putLong(lastMark.txnLogPosition);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("RollLog to persist last marked log : " + lastMark);
+ }
for(File dir: ledgerDirectories) {
File file = new File(dir, "lastMark");
try {
@@ -440,6 +575,12 @@ public class Bookie extends Thread {
}
}
}
+
+ /**
+ * Read last mark from lastMark file.
+ * The last mark should first be max journal log id,
+ * and then max log position in max journal log.
+ */
synchronized void readLog() {
byte buff[] = new byte[16];
ByteBuffer bb = ByteBuffer.wrap(buff);
@@ -454,56 +595,100 @@ public class Bookie extends Thread {
long p = bb.getLong();
if (i > txnLogId) {
txnLogId = i;
- }
- if (p > txnLogPosition) {
- txnLogPosition = p;
+ if(p > txnLogPosition) {
+ txnLogPosition = p;
+ }
}
} catch (IOException e) {
LOG.error("Problems reading from " + file + " (this is okay if it is the first time starting this bookie");
}
}
}
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+
+ sb.append("LastMark: logId - ").append(txnLogId)
+ .append(" , position - ").append(txnLogPosition);
+
+ return sb.toString();
+ }
}
private LastLogMark lastLogMark = new LastLogMark(0, 0);
+ LastLogMark getLastLogMark() {
+ return lastLogMark;
+ }
+
public boolean isRunning() {
return running;
}
+ /**
+ * A thread used for persisting journal entries to journal files.
+ *
+ * <p>
+ * Besides persisting journal entries, it also takes responsibility of
+ * rolling journal files when a journal file reaches journal file size
+ * limitation.
+ * </p>
+ * <p>
+ * During journal rolling, it first closes the writing journal, generates
+ * new journal file using current timestamp, and continue persistence logic.
+ * Those journals will be garbage collected in SyncThread.
+ * </p>
+ */
@Override
public void run() {
LinkedList<QueueEntry> toFlush = new LinkedList<QueueEntry>();
ByteBuffer lenBuff = ByteBuffer.allocate(4);
try {
- long logId = System.currentTimeMillis();
- FileChannel logFile = openChannel(logId);
- BufferedChannel bc = new BufferedChannel(logFile, 65536);
- zeros.clear();
- long nextPrealloc = preAllocSize;
+ long logId = 0;
+ FileChannel logFile = null;
+ BufferedChannel bc = null;
+ long nextPrealloc = 0;
long lastFlushPosition = 0;
- logFile.write(zeros, nextPrealloc);
- // TODO: Currently, when we roll over the journal logs, the older
- // ones are never garbage collected. We should remove a journal log
- // once all of its entries have been synced with the entry logs.
+ QueueEntry qe = null;
while (true) {
- QueueEntry qe = null;
- if (toFlush.isEmpty()) {
- qe = queue.take();
- } else {
- qe = queue.poll();
- if (qe == null || bc.position() > lastFlushPosition + 512*1024) {
- //logFile.force(false);
- bc.flush(true);
- lastFlushPosition = bc.position();
- lastLogMark.setLastLogMark(logId, lastFlushPosition);
- for (QueueEntry e : toFlush) {
- e.cb.writeComplete(0, e.ledgerId, e.entryId, null, e.ctx);
+ // new journal file to write
+ if (null == logFile) {
+ logId = System.currentTimeMillis();
+ logFile = openChannel(logId);
+ bc = new BufferedChannel(logFile, 65536);
+ zeros.clear();
+ nextPrealloc = preAllocSize;
+ lastFlushPosition = 0;
+ logFile.write(zeros, nextPrealloc);
+ }
+
+ if (qe == null) {
+ if (toFlush.isEmpty()) {
+ qe = queue.take();
+ } else {
+ qe = queue.poll();
+ if (qe == null || bc.position() > lastFlushPosition + 512*1024) {
+ //logFile.force(false);
+ bc.flush(true);
+ lastFlushPosition = bc.position();
+ lastLogMark.setLastLogMark(logId, lastFlushPosition);
+ for (QueueEntry e : toFlush) {
+ e.cb.writeComplete(0, e.ledgerId, e.entryId, null, e.ctx);
+ }
+ toFlush.clear();
+
+ // check wether journal file is over file limit
+ if (bc.position() > MAX_JOURNAL_SIZE) {
+ logFile.close();
+ logFile = null;
+ continue;
+ }
}
- toFlush.clear();
}
}
+
if (isZkExpired) {
LOG.warn("Exiting... zk client has expired.");
break;
@@ -526,6 +711,7 @@ public class Bookie extends Thread {
logFile.write(zeros, nextPrealloc);
}
toFlush.add(qe);
+ qe = null;
}
} catch (Exception e) {
LOG.fatal("Bookie thread exiting", e);
@@ -533,9 +719,18 @@ public class Bookie extends Thread {
}
private FileChannel openChannel(long logId) throws FileNotFoundException {
+ return openChannel(logId, 0);
+ }
+
+ private FileChannel openChannel(long logId, long position) throws FileNotFoundException {
FileChannel logFile = new RandomAccessFile(new File(journalDirectory,
Long.toHexString(logId) + ".txn"),
"rw").getChannel();
+ try {
+ logFile.position(position);
+ } catch (IOException e) {
+ LOG.fatal("Bookie journal file can seek to position :", e);
+ }
return logFile;
}
@@ -547,8 +742,7 @@ public class Bookie extends Thread {
if(zk != null) zk.close();
this.interrupt();
this.join();
- syncThread.running = false;
- syncThread.join();
+ syncThread.shutdown();
for(LedgerDescriptor d: ledgers.values()) {
d.close();
}
Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java?rev=1200806&r1=1200805&r2=1200806&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java (original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/EntryLogger.java Fri Nov 11 10:24:19 2011
@@ -482,6 +482,13 @@ public class EntryLogger {
gcThread.running = false;
gcThread.interrupt();
gcThread.join();
+ // since logChannel is buffered channel, do flush when shutting down
+ try {
+ flush();
+ } catch (IOException ie) {
+ // we have no idea how to avoid io exception during shutting down, so just ignore it
+ LOG.fatal("Error flush entry log during shutting down, which may cause entry log corrupted.", ie);
+ }
}
}
Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieJournalRollingTest.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieJournalRollingTest.java?rev=1200806&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieJournalRollingTest.java (added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieJournalRollingTest.java Fri Nov 11 10:24:19 2011
@@ -0,0 +1,222 @@
+package org.apache.bookkeeper.test;
+
+/*
+ *
+ * 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.
+ *
+ */
+import java.io.File;
+import java.util.Enumeration;
+import java.util.List;
+
+import org.apache.bookkeeper.bookie.Bookie;
+import org.apache.bookkeeper.bookie.Bookie.JournalIdFilter;
+import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.log4j.Logger;
+import org.junit.Assert;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * This class tests that bookie rolling journals
+ */
+public class BookieJournalRollingTest extends BaseTestCase {
+ static Logger LOG = Logger.getLogger(BookieJournalRollingTest.class);
+
+ DigestType digestType;
+
+ public BookieJournalRollingTest(DigestType digestType) {
+ super(3);
+ this.digestType = digestType;
+ }
+
+ @Before
+ @Override
+ public void setUp() throws Exception {
+ // Set up the configuration properties needed.
+ System.setProperty("journal_max_size_mb", "1");
+ System.setProperty("journal_max_backups", "1");
+ super.setUp();
+ }
+
+ @After
+ @Override
+ public void tearDown() throws Exception {
+ try {
+ super.tearDown();
+ } finally {
+ System.setProperty("journal_max_size_mb", "2048");
+ System.setProperty("journal_max_backups", "5");
+ }
+ }
+
+ /**
+ * Common method to create ledgers and write entries to them.
+ */
+ private LedgerHandle[] writeLedgerEntries(int numLedgers, int msgSize, int numMsgs) throws Exception {
+ // Create the ledgers
+ LedgerHandle[] lhs = new LedgerHandle[numLedgers];
+ long[] ledgerIds = new long[numLedgers];
+ for (int i = 0; i < numLedgers; i++) {
+ lhs[i] = bkc.createLedger(digestType, "".getBytes());
+ ledgerIds[i] = lhs[i].getId();
+ }
+
+ // Create a dummy message string to write as ledger entries
+ StringBuilder msgSB = new StringBuilder();
+ for (int i = 0; i < msgSize; i++) {
+ msgSB.append("a");
+ }
+ String msg = msgSB.toString();
+
+ // Write all of the entries for all of the ledgers
+ for (int i = 0; i < numMsgs; i++) {
+ for (int j = 0; j < numLedgers; j++) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(ledgerIds[j]).append('-').append(i).append('-')
+ .append(msg);
+ lhs[j].addEntry(sb.toString().getBytes());
+ }
+ }
+
+ // Return the ledger handles to the inserted ledgers and entries
+ return lhs;
+ }
+
+ private void validLedgerEntries(long[] ledgerIds, int msgSize, int numMsgs) throws Exception {
+ // Open the ledgers
+ LedgerHandle[] lhs = new LedgerHandle[ledgerIds.length];
+ for (int i = 0; i < lhs.length; i++) {
+ lhs[i] = bkc.openLedger(ledgerIds[i], digestType, "".getBytes());
+ }
+
+ StringBuilder msgSB = new StringBuilder();
+ for (int i = 0; i < msgSize; i++) {
+ msgSB.append("a");
+ }
+ String msg = msgSB.toString();
+
+ int numToRead = 10;
+ // read all of the entries for all the ledgers
+ for (int j = 0; j < lhs.length; j++) {
+ int start = 0;
+ int read = Math.min(numToRead, numMsgs - start);
+ int end = start + read - 1;
+ int entryId = 0;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Validating Entries of Ledger " + ledgerIds[j]);
+ }
+ while (start < numMsgs) {
+ Enumeration<LedgerEntry> seq = lhs[j].readEntries(start, end);
+ assertTrue("Enumeration of ledger entries has no element", seq.hasMoreElements() == true);
+ while (seq.hasMoreElements()) {
+ LedgerEntry e = seq.nextElement();
+ assertEquals(entryId, e.getEntryId());
+
+ StringBuilder sb = new StringBuilder();
+ sb.append(ledgerIds[j]).append('-').append(entryId).append('-')
+ .append(msg);
+ Assert.assertArrayEquals(sb.toString().getBytes(), e.getEntry());
+ entryId++;
+ }
+ assertEquals(entryId - 1, end);
+ start = end + 1;
+ read = Math.min(numToRead, numMsgs - start);
+ end = start + read - 1;
+ }
+ }
+ }
+
+ /**
+ * This test writes enough ledger entries to roll over the journals
+ *
+ * It will then keep only 1 journal file before last marked journal
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testJournalRolling() throws Exception {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Testing Journal Rolling");
+ }
+ // Write enough ledger entries so that we roll over journals
+ LedgerHandle[] lhs = writeLedgerEntries(4, 1024, 1024);
+ long[] ledgerIds = new long[lhs.length];
+ for (int i=0; i<lhs.length; i++) {
+ ledgerIds[i] = lhs[i].getId();
+ }
+
+ // Sleep for a while to ensure data are flushed
+ Thread.sleep(2000);
+
+ // verify that we only keep at most journal files
+ for (File journalDir : tmpDirs) {
+ List<Long> logs = Bookie.listJournalIds(journalDir, new JournalIdFilter() {
+ @Override
+ public boolean accept(long journalId) {
+ return true;
+ }
+ });
+ assertTrue(logs.size() <= 2);
+ }
+
+ // restart bookies
+ // ensure after restart we can read the entries since journals rolls
+ restartBookies();
+ validLedgerEntries(ledgerIds, 1024, 1024);
+ }
+
+ /**
+ * This test writes enough ledger entries to roll over the journals
+ * without sync up
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testJournalRollingWithoutSyncup() throws Exception {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Testing Journal Rolling without sync up");
+ }
+ // set flush interval to a large value
+ System.setProperty("flush_interval", "999999999");
+ try {
+ // restart bookies
+ restartBookies();
+
+ // Write enough ledger entries so that we roll over journals
+ LedgerHandle[] lhs = writeLedgerEntries(4, 1024, 1024);
+ long[] ledgerIds = new long[lhs.length];
+ for (int i=0; i<lhs.length; i++) {
+ ledgerIds[i] = lhs[i].getId();
+ }
+
+ // ledger indexes are not flushed
+ // and after bookies restarted, journals will be relayed
+ // ensure that we can still read the entries
+ restartBookies();
+ validLedgerEntries(ledgerIds, 1024, 1024);
+ } finally {
+ System.setProperty("flush_interval", "100");
+ }
+ }
+
+}
Modified: zookeeper/bookkeeper/trunk/doc/bookkeeperConfig.textile
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/doc/bookkeeperConfig.textile?rev=1200806&r1=1200805&r2=1200806&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/doc/bookkeeperConfig.textile (original)
+++ zookeeper/bookkeeper/trunk/doc/bookkeeperConfig.textile Fri Nov 11 10:24:19 2011
@@ -41,6 +41,35 @@ p. The configuration parameters, which c
p. Ideally, @/path_to_log_device/@ and @/path_to_ledger_device/@ are each in a different device.
+h1. Bookie Configuration
+
+p. Bookie server stores its data in multiple ledger directories and its journal files in a journal directory. Ideally, storing journal files in a separate directory than data files would increase throughput and decrease latency.
+
+h2. Journal Configuration
+
+p. Journal directory has one kind of files in it:
+
+* {timestamp}.txn - holds transactions executed in the bookie server.
+
+p. Before persisting ledger index and data to disk, a bookie ensures that the transaction that represents the update is written to a journal in non-volatile storage. A new journal file is created using current timestamp when a bookie starts or an old journal file reaches its maximum size.
+
+p. A bookie supports journal rolling to remove old journal files. In order to remove old journal files safely, bookie server records LastLogMark in Ledger Device, which indicates all updates (including index and data) before LastLogMark has been persisted to the Ledger Device.
+
+p. LastLogMark contains two parts:
+
+* LastLogId - indicates which journal file the transaction persisted.
+* LastLogPos - indicates the position the transaction persisted in LastLogId journal file.
+
+p. You may use following settings to further fine tune the behavior of your Bookie servers. Currently these configuration settings are set using Java system properties.
+
+* journal_max_size_mb
+** journal file size limitation. when a journal reaches this limitation, it will be closed and new journal file be created.
+
+* journal_max_backups
+** how many old journal files whose id is less than LastLogMark 's journal id.
+
+bq. NOTE: keeping number of old journal files would be useful for manually recovery in special case.
+
h1. ZooKeeper Metadata
-p. For BookKeeper, we require a ZooKeeper installation to store metadata, and to pass the list of ZooKeeper servers as parameter to the constructor of the BookKeeper class ( @org.apache.bookkeeper.client,BookKeeper@ ). To setup ZooKeeper, please check the "ZooKeeper documentation":index.html.
+p. For BookKeeper, we require a ZooKeeper installation to store metadata, and to pass the list of ZooKeeper servers as parameter to the constructor of the BookKeeper class ( @org.apache.bookkeeper.client,BookKeeper@ ). To setup ZooKeeper, please check the "ZooKeeper documentation":index.html.
\ No newline at end of file