core/src/main/java/org/apache/hadoop/chukwa/datacollection/writer/SeqFileWriter.java - chukwa - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.chukwa.datacollection.writer;


 import java.net.InetAddress;
 import java.net.URI;
 import java.net.UnknownHostException;
 import java.util.Calendar;
 import java.util.List;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.io.IOException;

 import org.apache.hadoop.chukwa.ChukwaArchiveKey;
 import org.apache.hadoop.chukwa.Chunk;
 import org.apache.hadoop.chukwa.ChunkImpl;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.SequenceFile;
 import org.apache.log4j.Logger;

 /**
  * This class <b>is</b> thread-safe -- rotate() and save() both synchronize on
  * this object.
  *
  */
 public class SeqFileWriter extends PipelineableWriter implements ChukwaWriter {
   static Logger log = Logger.getLogger(SeqFileWriter.class);
   private static boolean ENABLE_ROTATION_ON_CLOSE = true;

   protected int STAT_INTERVAL_SECONDS = 30;
   private int rotateInterval = 1000 * 60 * 5;
   private int offsetInterval = 1000 * 30;
   private boolean if_fixed_interval = false;
   static final int ACQ_WAIT_ON_TERM = 500; //ms to wait for lock on a SIGTERM before aborting

   public static final String STAT_PERIOD_OPT = "chukwaCollector.stats.period";
   public static final String ROTATE_INTERVAL_OPT = "chukwaCollector.rotateInterval";
   public static final String IF_FIXED_INTERVAL_OPT = "chukwaCollector.isFixedTimeRotatorScheme";
   public static final String FIXED_INTERVAL_OFFSET_OPT = "chukwaCollector.fixedTimeIntervalOffset";
   public static final String OUTPUT_DIR_OPT= "chukwaCollector.outputDir";
   public String localHostAddr = null;

   protected final Semaphore lock = new Semaphore(1, true);

   protected FileSystem fs = null;
   protected Configuration conf = null;

   protected String outputDir = null;
   private Calendar calendar = Calendar.getInstance();

   protected Path currentPath = null;
   protected String currentFileName = null;
   protected FSDataOutputStream currentOutputStr = null;
   protected SequenceFile.Writer seqFileWriter = null;

   protected long timePeriod = -1;
   protected long nextTimePeriodComputation = -1;

   protected Timer rotateTimer = null;
   protected Timer statTimer = null;

   protected volatile long dataSize = 0;
   protected volatile long bytesThisRotate = 0;
   protected volatile boolean isRunning = false;

   public SeqFileWriter() {
     try {
       localHostAddr = "_" + InetAddress.getLocalHost().getHostName() + "_";
     } catch (UnknownHostException e) {
       localHostAddr = "-NA-";
     }
   }

   public long getBytesWritten() {
     return dataSize;
   }

   public void init(Configuration conf) throws WriterException {
     outputDir = conf.get(OUTPUT_DIR_OPT, "/chukwa");

     this.conf = conf;

     rotateInterval = conf.getInt(ROTATE_INTERVAL_OPT,rotateInterval);
     if_fixed_interval = conf.getBoolean(IF_FIXED_INTERVAL_OPT,if_fixed_interval);
     offsetInterval = conf.getInt(FIXED_INTERVAL_OFFSET_OPT,offsetInterval);

     STAT_INTERVAL_SECONDS = conf.getInt(STAT_PERIOD_OPT, STAT_INTERVAL_SECONDS);

     // check if they've told us the file system to use
     String fsname = conf.get("writer.hdfs.filesystem");
     if (fsname == null || fsname.equals("")) {
       // otherwise try to get the filesystem from hadoop
       fsname = conf.get("fs.defaultFS");
     }

     log.info("rotateInterval is " + rotateInterval);
     if(if_fixed_interval)
       log.info("using fixed time interval scheme, " +
               "offsetInterval is " + offsetInterval);
     else
       log.info("not using fixed time interval scheme");
     log.info("outputDir is " + outputDir);
     log.info("fsname is " + fsname);
     log.info("filesystem type from core-default.xml is "
         + conf.get("fs.hdfs.impl"));

     if (fsname == null) {
       log.error("no filesystem name");
       throw new WriterException("no filesystem");
     }
     try {
       fs = FileSystem.get(new URI(fsname), conf);
       if (fs == null) {
         log.error("can't connect to HDFS.");
       }
     } catch (Throwable e) {
       log.error(
           "can't connect to HDFS, trying default file system instead (likely to be local)",
           e);
     }

     // Setup everything by rotating

     isRunning = true;
     rotate();

     statTimer = new Timer();
     statTimer.schedule(new StatReportingTask(), 1000,
         STAT_INTERVAL_SECONDS * 1000);

   }

   public class StatReportingTask extends TimerTask {
     private long lastTs = System.currentTimeMillis();

     public void run() {

       long time = System.currentTimeMillis();
       long currentDs = dataSize;
       dataSize = 0;

       long interval = time - lastTs;
       lastTs = time;

       long dataRate = 1000 * currentDs / interval; // kb/sec
       log.info("stat:datacollection.writer.hdfs dataSize=" + currentDs
           + " dataRate=" + dataRate);
     }

     public StatReportingTask() {}
   };

   void rotate() {
      if (rotateTimer != null) {
       rotateTimer.cancel();
     }

     if(!isRunning)
       return;

     calendar.setTimeInMillis(System.currentTimeMillis());

     String newName = new java.text.SimpleDateFormat("yyyyMMddHHmmssSSS")
         .format(calendar.getTime());
     newName += localHostAddr + new java.rmi.server.UID().toString();
     newName = newName.replace("-", "");
     newName = newName.replace(":", "");
     newName = newName.replace(".", "");
     newName = outputDir + "/" + newName.trim();

     try {
       lock.acquire();

       FSDataOutputStream previousOutputStr = currentOutputStr;
       Path previousPath = currentPath;
       String previousFileName = currentFileName;

       if (previousOutputStr != null) {
         boolean closed = false;
         try {
           log.info("closing sink file" + previousFileName);
           previousOutputStr.close();
           closed = true;
         }catch (Throwable e) {
           log.error("couldn't close file" + previousFileName, e);
           //we probably have an orphaned 0 byte file at this point due to an
           //intermitant HDFS outage. Once HDFS comes up again we'll be able to
           //close it, although it will be empty.
         }

         if (bytesThisRotate > 0) {
           if (closed) {
             log.info("rotating sink file " + previousPath);
             fs.rename(previousPath, new Path(previousFileName + ".done"));
           }
           else {
             log.warn(bytesThisRotate + " bytes potentially lost, since " +
                     previousPath + " could not be closed.");
           }
         } else {
           log.info("no chunks written to " + previousPath + ", deleting");
           fs.delete(previousPath, false);
         }
       }

       Path newOutputPath = new Path(newName + ".chukwa");
       FSDataOutputStream newOutputStr = fs.create(newOutputPath);
       // Uncompressed for now
       seqFileWriter = SequenceFile.createWriter(conf, newOutputStr,
           ChukwaArchiveKey.class, ChunkImpl.class,
           SequenceFile.CompressionType.NONE, null);

       // reset these once we know that seqFileWriter was created
       currentOutputStr = newOutputStr;
       currentPath = newOutputPath;
       currentFileName = newName;
       bytesThisRotate = 0;
     } catch (Throwable e) {
       log.warn("Got an exception trying to rotate. Will try again in " +
               rotateInterval/1000 + " seconds." ,e);
     } finally {
       lock.release();
     }

     // Schedule the next timer
     scheduleNextRotation();

   }

   /**
    * Schedules the rotate task using either a fixed time interval scheme or a
    * relative time interval scheme as specified by the
    * chukwaCollector.isFixedTimeRotatorScheme configuration parameter. If the
    * value of this parameter is true then next rotation will be scheduled at a
    * fixed offset from the current rotateInterval. This fixed offset is
    * provided by chukwaCollector.fixedTimeIntervalOffset configuration
    * parameter.
    */
   void scheduleNextRotation(){
     long delay = rotateInterval;
     if (if_fixed_interval) {
       long currentTime = System.currentTimeMillis();
       delay = getDelayForFixedInterval(currentTime, rotateInterval, offsetInterval);
     }
     rotateTimer = new Timer();
     rotateTimer.schedule(new TimerTask() {
       public void run() {
         rotate();
       }
     }, delay);
   }

   /**
    * Calculates delay for scheduling the next rotation in case of
    * FixedTimeRotatorScheme. This delay is the time difference between the
    * currentTimestamp (t1) and the next time the collector should rotate the
    * sequence files (t2). t2 is the time when the current rotateInterval ends
    * plus an offset (as set by chukwaCollector.FixedTimeIntervalOffset).
    * So, delay = t2 - t1
    *
    * @param currentTime - the current timestamp
    * @param rotateInterval - chukwaCollector.rotateInterval
    * @param offsetInterval - chukwaCollector.fixedTimeIntervalOffset
    * @return delay for scheduling next rotation
    */
   long getDelayForFixedInterval(long currentTime, long rotateInterval, long offsetInterval){
     // time since last rounded interval
     long remainder = (currentTime % rotateInterval);
     long prevRoundedInterval = currentTime - remainder;
     long nextRoundedInterval = prevRoundedInterval + rotateInterval;
     long delay = nextRoundedInterval - currentTime + offsetInterval;

     if (log.isInfoEnabled()) {
      log.info("currentTime="+currentTime+" prevRoundedInterval="+
              prevRoundedInterval+" nextRoundedInterval" +
             "="+nextRoundedInterval+" delay="+delay);
     }

     return delay;
   }


   protected void computeTimePeriod() {
     synchronized (calendar) {
       calendar.setTimeInMillis(System.currentTimeMillis());
       calendar.set(Calendar.MINUTE, 0);
       calendar.set(Calendar.SECOND, 0);
       calendar.set(Calendar.MILLISECOND, 0);
       timePeriod = calendar.getTimeInMillis();
       calendar.add(Calendar.HOUR, 1);
       nextTimePeriodComputation = calendar.getTimeInMillis();
     }
   }

   @Override
   public CommitStatus add(List<Chunk> chunks) throws WriterException {
     COMMIT_PENDING result = new COMMIT_PENDING(chunks.size());
     if (!isRunning) {
       log.info("Collector not ready");
       throw new WriterException("Collector not ready");
     }

     ChukwaArchiveKey archiveKey = new ChukwaArchiveKey();

     if (System.currentTimeMillis() >= nextTimePeriodComputation) {
       computeTimePeriod();
     }
     try {
       lock.acquire();
       for (Chunk chunk : chunks) {
         archiveKey.setTimePartition(timePeriod);
         archiveKey.setDataType(chunk.getDataType());
         archiveKey.setStreamName(chunk.getTags() + "/" + chunk.getSource()
             + "/" + chunk.getStreamName());
         archiveKey.setSeqId(chunk.getSeqID());

         seqFileWriter.append(archiveKey, chunk);

         // compute size for stats only if append succeeded. Note though that
         // seqFileWriter.append can continue taking data for quite some time
         // after HDFS goes down while the client is trying to reconnect. Hence
         // these stats might not reflect reality during an HDFS outage.
         dataSize += chunk.getData().length;
         bytesThisRotate += chunk.getData().length;

         String futureName = currentPath.getName().replace(".chukwa", ".done");
         result.addPend(futureName, currentOutputStr.getPos());

       }
     }
     catch (IOException e) {
       log.error("IOException when trying to write a chunk, Collector will return error and keep running.", e);
       return COMMIT_FAIL;
     }
     catch (Throwable e) {
       // We don't want to loose anything
       log.fatal("IOException when trying to write a chunk, Collector is going to exit!", e);
       isRunning = false;
     } finally {
       lock.release();
     }
     return result;
   }

   public void close() {

     isRunning = false;

     if (statTimer != null) {
       statTimer.cancel();
     }

     if (rotateTimer != null) {
       rotateTimer.cancel();
     }

     // If we are here it's either because of an HDFS exception
     // or Collector has received a kill -TERM
     boolean gotLock = false;
     try {
       gotLock = lock.tryAcquire(ACQ_WAIT_ON_TERM, TimeUnit.MILLISECONDS);
       if(gotLock) {

         if (this.currentOutputStr != null) {
           this.currentOutputStr.close();
         }
         if(ENABLE_ROTATION_ON_CLOSE)
           if(bytesThisRotate > 0)
             fs.rename(currentPath, new Path(currentFileName + ".done"));
           else
             fs.delete(currentPath, false);
       }
     } catch (Throwable e) {
      log.warn("cannot rename dataSink file:" + currentPath,e);
     } finally {
       if(gotLock)
         lock.release();
     }
   }

   public static void setEnableRotationOnClose(boolean b) {
     ENABLE_ROTATION_ON_CLOSE = b;
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.chukwa.datacollection.writer;


	import java.net.InetAddress;
	import java.net.URI;
	import java.net.UnknownHostException;
	import java.util.Calendar;
	import java.util.List;
	import java.util.Timer;
	import java.util.TimerTask;
	import java.util.concurrent.Semaphore;
	import java.util.concurrent.TimeUnit;
	import java.io.IOException;

	import org.apache.hadoop.chukwa.ChukwaArchiveKey;
	import org.apache.hadoop.chukwa.Chunk;
	import org.apache.hadoop.chukwa.ChunkImpl;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FSDataOutputStream;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.io.SequenceFile;
	import org.apache.log4j.Logger;

	/**
	* This class <b>is</b> thread-safe -- rotate() and save() both synchronize on
	* this object.
	*
	*/
	public class SeqFileWriter extends PipelineableWriter implements ChukwaWriter {
	static Logger log = Logger.getLogger(SeqFileWriter.class);
	private static boolean ENABLE_ROTATION_ON_CLOSE = true;

	protected int STAT_INTERVAL_SECONDS = 30;
	private int rotateInterval = 1000 * 60 * 5;
	private int offsetInterval = 1000 * 30;
	private boolean if_fixed_interval = false;
	static final int ACQ_WAIT_ON_TERM = 500; //ms to wait for lock on a SIGTERM before aborting

	public static final String STAT_PERIOD_OPT = "chukwaCollector.stats.period";
	public static final String ROTATE_INTERVAL_OPT = "chukwaCollector.rotateInterval";
	public static final String IF_FIXED_INTERVAL_OPT = "chukwaCollector.isFixedTimeRotatorScheme";
	public static final String FIXED_INTERVAL_OFFSET_OPT = "chukwaCollector.fixedTimeIntervalOffset";
	public static final String OUTPUT_DIR_OPT= "chukwaCollector.outputDir";
	public String localHostAddr = null;

	protected final Semaphore lock = new Semaphore(1, true);

	protected FileSystem fs = null;
	protected Configuration conf = null;

	protected String outputDir = null;
	private Calendar calendar = Calendar.getInstance();

	protected Path currentPath = null;
	protected String currentFileName = null;
	protected FSDataOutputStream currentOutputStr = null;
	protected SequenceFile.Writer seqFileWriter = null;

	protected long timePeriod = -1;
	protected long nextTimePeriodComputation = -1;

	protected Timer rotateTimer = null;
	protected Timer statTimer = null;

	protected volatile long dataSize = 0;
	protected volatile long bytesThisRotate = 0;
	protected volatile boolean isRunning = false;

	public SeqFileWriter() {
	try {
	localHostAddr = "_" + InetAddress.getLocalHost().getHostName() + "_";
	} catch (UnknownHostException e) {
	localHostAddr = "-NA-";
	}
	}

	public long getBytesWritten() {
	return dataSize;
	}

	public void init(Configuration conf) throws WriterException {
	outputDir = conf.get(OUTPUT_DIR_OPT, "/chukwa");

	this.conf = conf;

	rotateInterval = conf.getInt(ROTATE_INTERVAL_OPT,rotateInterval);
	if_fixed_interval = conf.getBoolean(IF_FIXED_INTERVAL_OPT,if_fixed_interval);
	offsetInterval = conf.getInt(FIXED_INTERVAL_OFFSET_OPT,offsetInterval);

	STAT_INTERVAL_SECONDS = conf.getInt(STAT_PERIOD_OPT, STAT_INTERVAL_SECONDS);

	// check if they've told us the file system to use
	String fsname = conf.get("writer.hdfs.filesystem");
	if (fsname == null \|\| fsname.equals("")) {
	// otherwise try to get the filesystem from hadoop
	fsname = conf.get("fs.defaultFS");
	}

	log.info("rotateInterval is " + rotateInterval);
	if(if_fixed_interval)
	log.info("using fixed time interval scheme, " +
	"offsetInterval is " + offsetInterval);
	else
	log.info("not using fixed time interval scheme");
	log.info("outputDir is " + outputDir);
	log.info("fsname is " + fsname);
	log.info("filesystem type from core-default.xml is "
	+ conf.get("fs.hdfs.impl"));

	if (fsname == null) {
	log.error("no filesystem name");
	throw new WriterException("no filesystem");
	}
	try {
	fs = FileSystem.get(new URI(fsname), conf);
	if (fs == null) {
	log.error("can't connect to HDFS.");
	}
	} catch (Throwable e) {
	log.error(
	"can't connect to HDFS, trying default file system instead (likely to be local)",
	e);
	}

	// Setup everything by rotating

	isRunning = true;
	rotate();

	statTimer = new Timer();
	statTimer.schedule(new StatReportingTask(), 1000,
	STAT_INTERVAL_SECONDS * 1000);

	}

	public class StatReportingTask extends TimerTask {
	private long lastTs = System.currentTimeMillis();

	public void run() {

	long time = System.currentTimeMillis();
	long currentDs = dataSize;
	dataSize = 0;

	long interval = time - lastTs;
	lastTs = time;

	long dataRate = 1000 * currentDs / interval; // kb/sec
	log.info("stat:datacollection.writer.hdfs dataSize=" + currentDs
	+ " dataRate=" + dataRate);
	}

	public StatReportingTask() {}
	};

	void rotate() {
	if (rotateTimer != null) {
	rotateTimer.cancel();
	}

	if(!isRunning)
	return;

	calendar.setTimeInMillis(System.currentTimeMillis());

	String newName = new java.text.SimpleDateFormat("yyyyMMddHHmmssSSS")
	.format(calendar.getTime());
	newName += localHostAddr + new java.rmi.server.UID().toString();
	newName = newName.replace("-", "");
	newName = newName.replace(":", "");
	newName = newName.replace(".", "");
	newName = outputDir + "/" + newName.trim();

	try {
	lock.acquire();

	FSDataOutputStream previousOutputStr = currentOutputStr;
	Path previousPath = currentPath;
	String previousFileName = currentFileName;

	if (previousOutputStr != null) {
	boolean closed = false;
	try {
	log.info("closing sink file" + previousFileName);
	previousOutputStr.close();
	closed = true;
	}catch (Throwable e) {
	log.error("couldn't close file" + previousFileName, e);
	//we probably have an orphaned 0 byte file at this point due to an
	//intermitant HDFS outage. Once HDFS comes up again we'll be able to
	//close it, although it will be empty.
	}

	if (bytesThisRotate > 0) {
	if (closed) {
	log.info("rotating sink file " + previousPath);
	fs.rename(previousPath, new Path(previousFileName + ".done"));
	}
	else {
	log.warn(bytesThisRotate + " bytes potentially lost, since " +
	previousPath + " could not be closed.");
	}
	} else {
	log.info("no chunks written to " + previousPath + ", deleting");
	fs.delete(previousPath, false);
	}
	}

	Path newOutputPath = new Path(newName + ".chukwa");
	FSDataOutputStream newOutputStr = fs.create(newOutputPath);
	// Uncompressed for now
	seqFileWriter = SequenceFile.createWriter(conf, newOutputStr,
	ChukwaArchiveKey.class, ChunkImpl.class,
	SequenceFile.CompressionType.NONE, null);

	// reset these once we know that seqFileWriter was created
	currentOutputStr = newOutputStr;
	currentPath = newOutputPath;
	currentFileName = newName;
	bytesThisRotate = 0;
	} catch (Throwable e) {
	log.warn("Got an exception trying to rotate. Will try again in " +
	rotateInterval/1000 + " seconds." ,e);
	} finally {
	lock.release();
	}

	// Schedule the next timer
	scheduleNextRotation();

	}

	/**
	* Schedules the rotate task using either a fixed time interval scheme or a
	* relative time interval scheme as specified by the
	* chukwaCollector.isFixedTimeRotatorScheme configuration parameter. If the
	* value of this parameter is true then next rotation will be scheduled at a
	* fixed offset from the current rotateInterval. This fixed offset is
	* provided by chukwaCollector.fixedTimeIntervalOffset configuration
	* parameter.
	*/
	void scheduleNextRotation(){
	long delay = rotateInterval;
	if (if_fixed_interval) {
	long currentTime = System.currentTimeMillis();
	delay = getDelayForFixedInterval(currentTime, rotateInterval, offsetInterval);
	}
	rotateTimer = new Timer();
	rotateTimer.schedule(new TimerTask() {
	public void run() {
	rotate();
	}
	}, delay);
	}

	/**
	* Calculates delay for scheduling the next rotation in case of
	* FixedTimeRotatorScheme. This delay is the time difference between the
	* currentTimestamp (t1) and the next time the collector should rotate the
	* sequence files (t2). t2 is the time when the current rotateInterval ends
	* plus an offset (as set by chukwaCollector.FixedTimeIntervalOffset).
	* So, delay = t2 - t1
	*
	* @param currentTime - the current timestamp
	* @param rotateInterval - chukwaCollector.rotateInterval
	* @param offsetInterval - chukwaCollector.fixedTimeIntervalOffset
	* @return delay for scheduling next rotation
	*/
	long getDelayForFixedInterval(long currentTime, long rotateInterval, long offsetInterval){
	// time since last rounded interval
	long remainder = (currentTime % rotateInterval);
	long prevRoundedInterval = currentTime - remainder;
	long nextRoundedInterval = prevRoundedInterval + rotateInterval;
	long delay = nextRoundedInterval - currentTime + offsetInterval;

	if (log.isInfoEnabled()) {
	log.info("currentTime="+currentTime+" prevRoundedInterval="+
	prevRoundedInterval+" nextRoundedInterval" +
	"="+nextRoundedInterval+" delay="+delay);
	}

	return delay;
	}


	protected void computeTimePeriod() {
	synchronized (calendar) {
	calendar.setTimeInMillis(System.currentTimeMillis());
	calendar.set(Calendar.MINUTE, 0);
	calendar.set(Calendar.SECOND, 0);
	calendar.set(Calendar.MILLISECOND, 0);
	timePeriod = calendar.getTimeInMillis();
	calendar.add(Calendar.HOUR, 1);
	nextTimePeriodComputation = calendar.getTimeInMillis();
	}
	}

	@Override
	public CommitStatus add(List<Chunk> chunks) throws WriterException {
	COMMIT_PENDING result = new COMMIT_PENDING(chunks.size());
	if (!isRunning) {
	log.info("Collector not ready");
	throw new WriterException("Collector not ready");
	}

	ChukwaArchiveKey archiveKey = new ChukwaArchiveKey();

	if (System.currentTimeMillis() >= nextTimePeriodComputation) {
	computeTimePeriod();
	}
	try {
	lock.acquire();
	for (Chunk chunk : chunks) {
	archiveKey.setTimePartition(timePeriod);
	archiveKey.setDataType(chunk.getDataType());
	archiveKey.setStreamName(chunk.getTags() + "/" + chunk.getSource()
	+ "/" + chunk.getStreamName());
	archiveKey.setSeqId(chunk.getSeqID());

	seqFileWriter.append(archiveKey, chunk);

	// compute size for stats only if append succeeded. Note though that
	// seqFileWriter.append can continue taking data for quite some time
	// after HDFS goes down while the client is trying to reconnect. Hence
	// these stats might not reflect reality during an HDFS outage.
	dataSize += chunk.getData().length;
	bytesThisRotate += chunk.getData().length;

	String futureName = currentPath.getName().replace(".chukwa", ".done");
	result.addPend(futureName, currentOutputStr.getPos());

	}
	}
	catch (IOException e) {
	log.error("IOException when trying to write a chunk, Collector will return error and keep running.", e);
	return COMMIT_FAIL;
	}
	catch (Throwable e) {
	// We don't want to loose anything
	log.fatal("IOException when trying to write a chunk, Collector is going to exit!", e);
	isRunning = false;
	} finally {
	lock.release();
	}
	return result;
	}

	public void close() {

	isRunning = false;

	if (statTimer != null) {
	statTimer.cancel();
	}

	if (rotateTimer != null) {
	rotateTimer.cancel();
	}

	// If we are here it's either because of an HDFS exception
	// or Collector has received a kill -TERM
	boolean gotLock = false;
	try {
	gotLock = lock.tryAcquire(ACQ_WAIT_ON_TERM, TimeUnit.MILLISECONDS);
	if(gotLock) {

	if (this.currentOutputStr != null) {
	this.currentOutputStr.close();
	}
	if(ENABLE_ROTATION_ON_CLOSE)
	if(bytesThisRotate > 0)
	fs.rename(currentPath, new Path(currentFileName + ".done"));
	else
	fs.delete(currentPath, false);
	}
	} catch (Throwable e) {
	log.warn("cannot rename dataSink file:" + currentPath,e);
	} finally {
	if(gotLock)
	lock.release();
	}
	}

	public static void setEnableRotationOnClose(boolean b) {
	ENABLE_ROTATION_ON_CLOSE = b;
	}

	}