BKLogWriteHandler(ZKLogMetadataForWriter logMetadata,DistributedLogConfiguration conf,ZooKeeperClientBuilder zkcBuilder,BookKeeperClientBuilder bkcBuilder,LogSegmentMetadataStore metadataStore,OrderedScheduler scheduler,LedgerAllocator allocator,StatsLogger statsLogger,StatsLogger perLogStatsLogger,AlertStatsLogger alertStatsLogger,String clientId,int regionId,PermitLimiter writeLimiter,FeatureProvider featureProvider,DynamicDistributedLogConfiguration dynConf,DistributedLock lock){
  super(logMetadata,conf,zkcBuilder,bkcBuilder,metadataStore,scheduler,statsLogger,alertStatsLogger,null,WRITE_HANDLE_FILTER,clientId);
  this.perLogStatsLogger=perLogStatsLogger;
  this.writeLimiter=writeLimiter;
  this.featureProvider=featureProvider;
  this.dynConf=dynConf;
  this.ledgerAllocator=allocator;
  this.lock=lock;
  this.metadataUpdater=LogSegmentMetadataStoreUpdater.createMetadataUpdater(conf,metadataStore);
  ensembleSize=conf.getEnsembleSize();
  if (ensembleSize < conf.getWriteQuorumSize()) {
    writeQuorumSize=ensembleSize;
    LOG.warn("Setting write quorum size {} greater than ensemble size {}",conf.getWriteQuorumSize(),ensembleSize);
  }
 else {
    writeQuorumSize=conf.getWriteQuorumSize();
  }
  if (writeQuorumSize < conf.getAckQuorumSize()) {
    ackQuorumSize=writeQuorumSize;
    LOG.warn("Setting write ack quorum size {} greater than write quorum size {}",conf.getAckQuorumSize(),writeQuorumSize);
  }
 else {
    ackQuorumSize=conf.getAckQuorumSize();
  }
  if (conf.getEncodeRegionIDInLogSegmentMetadata()) {
    this.regionId=regionId;
  }
 else {
    this.regionId=DistributedLogConstants.LOCAL_REGION_ID;
  }
  this.sanityCheckTxnId=conf.getSanityCheckTxnID();
  this.validateLogSegmentSequenceNumber=conf.isLogSegmentSequenceNumberValidationEnabled();
  maxLogSegmentSequenceNo=new MaxLogSegmentSequenceNo(logMetadata.getMaxLSSNData());
  inprogressLSSNs=new LinkedList<Long>();
  maxTxId=new MaxTxId(zooKeeperClient,logMetadata.getMaxTxIdPath(),conf.getSanityCheckTxnID(),logMetadata.getMaxTxIdData());
  scheduleGetLedgersTask(false,true);
  setLastLedgerRollingTimeMillis(Utils.nowInMillis());
  if (conf.getLogSegmentRollingIntervalMinutes() > 0) {
    rollingPolicy=new TimeBasedRollingPolicy(conf.getLogSegmentRollingIntervalMinutes() * 60 * 1000);
  }
 else {
    rollingPolicy=new SizeBasedRollingPolicy(conf.getMaxLogSegmentBytes());
  }
  StatsLogger segmentsStatsLogger=statsLogger.scope("segments");
  openOpStats=segmentsStatsLogger.getOpStatsLogger("open");
  closeOpStats=segmentsStatsLogger.getOpStatsLogger("close");
  recoverOpStats=segmentsStatsLogger.getOpStatsLogger("recover");
  deleteOpStats=segmentsStatsLogger.getOpStatsLogger("delete");
}

BKLogWriteHandler ( ZKLogMetadataForWriter logMetadata , DistributedLogConfiguration conf , ZooKeeperClientBuilder zkcBuilder , BookKeeperClientBuilder bkcBuilder , LogSegmentMetadataStore metadataStore , OrderedScheduler scheduler , LedgerAllocator allocator , StatsLogger statsLogger , StatsLogger perLogStatsLogger , AlertStatsLogger alertStatsLogger , String clientId , int regionId , PermitLimiter writeLimiter , FeatureProvider featureProvider , DynamicDistributedLogConfiguration dynConf , DistributedLock lock ) { super ( logMetadata , conf , zkcBuilder , bkcBuilder , metadataStore , scheduler , statsLogger , alertStatsLogger , null , WRITE_HANDLE_FILTER , clientId ) ; this . perLogStatsLogger = perLogStatsLogger ; this . writeLimiter = writeLimiter ; this . featureProvider = featureProvider ; this . dynConf = dynConf ; this . ledgerAllocator = allocator ; this . lock = lock ; this . metadataUpdater = LogSegmentMetadataStoreUpdater . createMetadataUpdater ( conf , metadataStore ) ; ensembleSize = conf . getEnsembleSize ( ) ; if ( ensembleSize < conf . getWriteQuorumSize ( ) ) { writeQuorumSize = ensembleSize ; LOG . warn ( "Setting write quorum size {} greater than ensemble size {}" , conf . getWriteQuorumSize ( ) , ensembleSize ) ; } else { writeQuorumSize = conf . getWriteQuorumSize ( ) ; } if ( writeQuorumSize < conf . getAckQuorumSize ( ) ) { ackQuorumSize = writeQuorumSize ; LOG . warn ( "Setting write ack quorum size {} greater than write quorum size {}" , conf . getAckQuorumSize ( ) , writeQuorumSize ) ; } else { ackQuorumSize = conf . getAckQuorumSize ( ) ; } if ( conf . getEncodeRegionIDInLogSegmentMetadata ( ) ) { this . regionId = regionId ; } else { this . regionId = DistributedLogConstants . LOCAL_REGION_ID ; } this . sanityCheckTxnId = conf . getSanityCheckTxnID ( ) ; this . validateLogSegmentSequenceNumber = conf . isLogSegmentSequenceNumberValidationEnabled ( ) ; maxLogSegmentSequenceNo = new MaxLogSegmentSequenceNo ( logMetadata . getMaxLSSNData ( ) ) ; inprogressLSSNs = new LinkedList < Long > ( ) ; maxTxId = new MaxTxId ( zooKeeperClient , logMetadata . getMaxTxIdPath ( ) , conf . getSanityCheckTxnID ( ) , logMetadata . getMaxTxIdData ( ) ) ; scheduleGetLedgersTask ( false , true ) ; setLastLedgerRollingTimeMillis ( Utils . nowInMillis ( ) ) ; if ( conf . getLogSegmentRollingIntervalMinutes ( ) > 0 ) { rollingPolicy = new TimeBasedRollingPolicy ( conf . getLogSegmentRollingIntervalMinutes ( ) * 60 * 1000 ) ; } else { rollingPolicy = new SizeBasedRollingPolicy ( conf . getMaxLogSegmentBytes ( ) ) ; } StatsLogger segmentsStatsLogger = statsLogger . scope ( "segments" ) ; openOpStats = segmentsStatsLogger . getOpStatsLogger ( "open" ) ; closeOpStats = segmentsStatsLogger . getOpStatsLogger ( "close" ) ; recoverOpStats = segmentsStatsLogger . getOpStatsLogger ( "recover" ) ; deleteOpStats = segmentsStatsLogger . getOpStatsLogger ( "delete" ) ; }
construct a bookkeeper journal manager .
