You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by mp...@apache.org on 2012/06/26 09:02:52 UTC
svn commit: r1353836 - in
/incubator/flume/trunk/flume-ng-channels/flume-file-channel/src:
main/java/org/apache/flume/channel/file/
test/java/org/apache/flume/channel/file/
Author: mpercy
Date: Tue Jun 26 07:02:49 2012
New Revision: 1353836
URL: http://svn.apache.org/viewvc?rev=1353836&view=rev
Log:
FLUME-1314. Log file size violation prevents FileChannel startup.
(Arvind Prabhakar via Mike Percy)
Modified:
incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannelConfiguration.java
incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java
incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java
Modified: incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannelConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannelConfiguration.java?rev=1353836&r1=1353835&r2=1353836&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannelConfiguration.java (original)
+++ incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/FileChannelConfiguration.java Tue Jun 26 07:02:49 2012
@@ -40,10 +40,11 @@ public class FileChannelConfiguration {
public static final String CHECKPOINT_INTERVAL = "checkpointInterval";
public static final long DEFAULT_CHECKPOINT_INTERVAL = 30L * 1000L;
/**
- * Max file size for data files, cannot exceed the default. Default: 2GB
+ * Max file size for data files, cannot exceed the default. Default~ 1.5GB
*/
public static final String MAX_FILE_SIZE = "maxFileSize";
- public static final long DEFAULT_MAX_FILE_SIZE = LogFile.MAX_FILE_SIZE;
+ public static final long DEFAULT_MAX_FILE_SIZE =
+ Integer.MAX_VALUE - (500L * 1024L * 1024L); // ~1.52 G
/**
* Maximum capacity of the channel. This number needs to be configured
* in line with -XX:MaxDirectMemorySize. {@link FileChannel}
@@ -63,5 +64,5 @@ public class FileChannelConfiguration {
* checkpoint is enqueued or in progress.
*/
public static final String LOG_WRITE_TIMEOUT = "write-timeout";
- public static final int DEFAULT_WRITE_TIMEOUT = 3;
+ public static final int DEFAULT_WRITE_TIMEOUT = 10;
}
Modified: incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java?rev=1353836&r1=1353835&r2=1353836&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java (original)
+++ incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/LogFile.java Tue Jun 26 07:02:49 2012
@@ -50,8 +50,6 @@ class LogFile {
*/
private static final ByteBuffer FILL = DirectMemoryUtils.
allocate(1024 * 1024); // preallocation, 1MB
- public static final long MAX_FILE_SIZE =
- Integer.MAX_VALUE - (1024L * 1024L);
private static final byte OP_RECORD = Byte.MAX_VALUE;
private static final byte OP_EOF = Byte.MIN_VALUE;
@@ -77,7 +75,8 @@ class LogFile {
Writer(File file, int logFileID, long maxFileSize) throws IOException {
this.file = file;
fileID = logFileID;
- this.maxFileSize = Math.min(maxFileSize, MAX_FILE_SIZE);
+ this.maxFileSize = Math.min(maxFileSize,
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE);
writeFileHandle = new RandomAccessFile(file, "rw");
writeFileHandle.writeInt(VERSION);
writeFileHandle.writeInt(fileID);
@@ -332,8 +331,11 @@ class LogFile {
Pair<Integer, TransactionEventRecord> next() throws IOException {
try {
long position = fileChannel.position();
- Preconditions.checkState(position < MAX_FILE_SIZE,
- String.valueOf(position));
+ if (position > FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE) {
+ LOG.warn("File position exceeds the threshold: "
+ + FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE
+ + ", position: " + position);
+ }
int offset = (int) position;
byte operation = fileHandle.readByte();
if(operation != OP_RECORD) {
Modified: incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java?rev=1353836&r1=1353835&r2=1353836&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java (original)
+++ incubator/flume/trunk/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/TestLog.java Tue Jun 26 07:02:49 2012
@@ -114,7 +114,8 @@ public class TestLog {
log.commitPut(transactionID);
log.close();
log = new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
CAPACITY).setCheckpointDir(checkpointDir).setLogDirs(
dataDirs).build();
log.replay();
@@ -133,7 +134,8 @@ public class TestLog {
log.rollback(transactionID); // rolled back so it should not be replayed
log.close();
log = new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
CAPACITY).setCheckpointDir(checkpointDir).setLogDirs(
dataDirs).build();
log.replay();
@@ -156,7 +158,8 @@ public class TestLog {
log.commitTake(takeTransactionID);
log.close();
new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
1).setCheckpointDir(checkpointDir).setLogDirs(dataDirs).build();
log.replay();
FlumeEventQueue queue = log.getFlumeEventQueue();
@@ -178,7 +181,8 @@ public class TestLog {
log.rollback(takeTransactionID);
log.close();
new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
1).setCheckpointDir(checkpointDir).setLogDirs(dataDirs).build();
log.replay();
takeAndVerify(eventPointerIn, eventIn);
@@ -190,7 +194,8 @@ public class TestLog {
log.commitPut(putTransactionID);
log.close();
new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
1).setCheckpointDir(checkpointDir).setLogDirs(dataDirs).build();
log.replay();
FlumeEventQueue queue = log.getFlumeEventQueue();
@@ -204,7 +209,8 @@ public class TestLog {
log.commitTake(putTransactionID);
log.close();
new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
1).setCheckpointDir(checkpointDir).setLogDirs(dataDirs).build();
log.replay();
FlumeEventQueue queue = log.getFlumeEventQueue();
@@ -218,7 +224,8 @@ public class TestLog {
log.rollback(putTransactionID);
log.close();
new Log.Builder().setCheckpointInterval(
- Long.MAX_VALUE).setMaxFileSize(LogFile.MAX_FILE_SIZE).setQueueSize(
+ Long.MAX_VALUE).setMaxFileSize(
+ FileChannelConfiguration.DEFAULT_MAX_FILE_SIZE).setQueueSize(
1).setCheckpointDir(checkpointDir).setLogDirs(dataDirs).build();
log.replay();
FlumeEventQueue queue = log.getFlumeEventQueue();