public KafkaReader(TaskAttemptContext context,EtlRequest request,int clientTimeout,int fetchBufferSize) throws Exception {
  this.fetchBufferSize=fetchBufferSize;
  this.context=context;
  log.info("bufferSize=" + fetchBufferSize);
  log.info("timeout=" + clientTimeout);
  kafkaRequest=request;
  beginOffset=request.getOffset();
  currentOffset=request.getOffset();
  lastOffset=request.getLastOffset();
  currentCount=0;
  totalFetchTime=0;
  URI uri=kafkaRequest.getURI();
  simpleConsumer=new SimpleConsumer(uri.getHost(),uri.getPort(),CamusJob.getKafkaTimeoutValue(context),CamusJob.getKafkaBufferSize(context),CamusJob.getKafkaClientName(context));
  log.info("Connected to leader " + uri + " beginning reading at offset "+ beginOffset+ " latest offset="+ lastOffset);
  fetch();
}

public KafkaReader ( TaskAttemptContext context , EtlRequest request , int clientTimeout , int fetchBufferSize ) throws Exception { this . fetchBufferSize = fetchBufferSize ; this . context = context ; log . info ( "bufferSize=" + fetchBufferSize ) ; log . info ( "timeout=" + clientTimeout ) ; kafkaRequest = request ; beginOffset = request . getOffset ( ) ; currentOffset = request . getOffset ( ) ; lastOffset = request . getLastOffset ( ) ; currentCount = 0 ; totalFetchTime = 0 ; URI uri = kafkaRequest . getURI ( ) ; simpleConsumer = new SimpleConsumer ( uri . getHost ( ) , uri . getPort ( ) , CamusJob . getKafkaTimeoutValue ( context ) , CamusJob . getKafkaBufferSize ( context ) , CamusJob . getKafkaClientName ( context ) ) ; log . info ( "Connected to leader " + uri + " beginning reading at offset " + beginOffset + " latest offset=" + lastOffset ) ; fetch ( ) ; }
construct using the json representation of the kafka request
