private void processFile() {
		int lineNumber = 0;
		
		log.trace("Hosts: " + this.client.getNodeNames());
		long start = System.currentTimeMillis();
		try{
			BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(this.fileName), "UTF8"));
			log.debug("Reading file:  " + Utils.getFileName(fileName));
			while(br.ready())
			{
				String line;
				
				//skip reading 1st line of data file
				if (params.ignoreFirstLine){
					lineNumber++;	
					String metaColumn = br.readLine();
					counters.write.recordTotal = counters.write.recordTotal - metaColumn.length();
				}
				AsWriterTask task = null;
				while((line = br.readLine()) != null){
					lineNumber++;
					log.trace("Read line " + lineNumber + " from file " + Utils.getFileName(fileName));
					
					// Throttle the read to write ratio
					while((counters.write.readCount.get() - (counters.write.writeCount.get() + counters.write.writeErrors.get())) > rwThrottle) {
						Thread.sleep(20);
					}
					if(params.fileType.equalsIgnoreCase(Constants.CSV_FILE)){
						//add 2 to handle different file size in different platform.
						task = new AsWriterTask(fileName, lineNumber, (line.length() + 1), Parser.getCSVRawColumns(line, params.delimiter), metadataColumnDefs, binColumnDefs, this.client, params, counters);
						counters.write.readerProcessed.incrementAndGet();
					} else {
						log.error("File format not supported");
					}
					log.trace("Submitting line " + lineNumber + " in file " + Utils.getFileName(fileName));
					writerPool.submit(task);
					
					counters.write.readCount.incrementAndGet();
				}
			}
			
			br.close();
			
		} catch (IOException e) {
			counters.write.readErrors.incrementAndGet();
			log.error("Error processing file: " + Utils.getFileName(fileName)+": Line: " + lineNumber);
			if(log.isDebugEnabled()){
				e.printStackTrace();
			}
		} catch (Exception e) {
			counters.write.readErrors.incrementAndGet();
			log.error("Error processing file: " + Utils.getFileName(fileName)+": Line: "+ lineNumber);
			if(log.isDebugEnabled()){
				e.printStackTrace();
			}
		} 
		long stop = System.currentTimeMillis();
		log.info(String.format("Reader completed %d-lines in %.3fsec, From file: %s", 
				lineNumber, (float)(stop-start)/1000, Utils.getFileName(fileName)));
	}