import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public ListenableFuture<PrepareRecoveryResponseProto> prepareRecovery(
    final long segmentTxId) {
  return singleThreadExecutor.submit(new Callable<PrepareRecoveryResponseProto>() {
    @Override
    public PrepareRecoveryResponseProto call() throws IOException {
      if (!hasHttpServerEndPoint()) {
        // force an RPC call so we know what the HTTP port should be if it
        // haven't done so.
        GetJournalStateResponseProto ret = getProxy().getJournalState(
            journalId);
        constructHttpServerURI(ret);
      }
      return getProxy().prepareRecovery(createReqInfo(), segmentTxId);
    }
  });
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public ListenableFuture<PrepareRecoveryResponseProto> prepareRecovery(
    final long segmentTxId) {
  return executor.submit(new Callable<PrepareRecoveryResponseProto>() {
    @Override
    public PrepareRecoveryResponseProto call() throws IOException {
      if (!hasHttpServerEndPoint()) {
        // force an RPC call so we know what the HTTP port should be if it
        // haven't done so.
        GetJournalStateResponseProto ret = getProxy().getJournalState(
            journalId);
        constructHttpServerURI(ret);
      }
      return getProxy().prepareRecovery(createReqInfo(), segmentTxId);
    }
  });
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
public QuorumCall<AsyncLogger, GetJournalStateResponseProto> getJournalState() {
  Map<AsyncLogger, ListenableFuture<GetJournalStateResponseProto>> calls =
      Maps.newHashMap();
  for (AsyncLogger logger : loggers) {
    calls.put(logger, logger.getJournalState());
  }
  return QuorumCall.create(calls);    
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public ListenableFuture<GetJournalStateResponseProto> getJournalState() {
  return singleThreadExecutor.submit(new Callable<GetJournalStateResponseProto>() {
    @Override
    public GetJournalStateResponseProto call() throws IOException {
      GetJournalStateResponseProto ret =
          getProxy().getJournalState(journalId);
      constructHttpServerURI(ret);
      return ret;
    }
  });
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
private void constructHttpServerURI(GetJournalStateResponseProto ret) {
  if (ret.hasFromURL()) {
    URI uri = URI.create(ret.getFromURL());
    httpServerURL = getHttpServerURI(uri.getScheme(), uri.getPort());
  } else {
    httpServerURL = getHttpServerURI("http", ret.getHttpPort());;
  }
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@SuppressWarnings("deprecation")
@Override
public GetJournalStateResponseProto getJournalState(String journalId)
      throws IOException {
  long epoch = jn.getOrCreateJournal(journalId).getLastPromisedEpoch(); 
  return GetJournalStateResponseProto.newBuilder()
      .setLastPromisedEpoch(epoch)
      .setHttpPort(jn.getBoundHttpAddress().getPort())
      .setFromURL(jn.getHttpServerURI())
      .build();
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public GetJournalStateResponseProto getJournalState(RpcController controller,
    GetJournalStateRequestProto request) throws ServiceException {
  try {
    return impl.getJournalState(
        convert(request.getJid()));
  } catch (IOException ioe) {
    throw new ServiceException(ioe);
  }
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public GetJournalStateResponseProto getJournalState(String jid)
    throws IOException {
  try {
    GetJournalStateRequestProto req = GetJournalStateRequestProto.newBuilder()
        .setJid(convertJournalId(jid))
        .build();
    return rpcProxy.getJournalState(NULL_CONTROLLER, req);
  } catch (ServiceException e) {
    throw ProtobufHelper.getRemoteException(e);
  }
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  spyLoggers = ImmutableList.of(
      mockLogger(),
      mockLogger(),
      mockLogger());

  qjm = new QuorumJournalManager(conf, new URI("qjournal://host/jid"), FAKE_NSINFO) {
    @Override
    protected List<AsyncLogger> createLoggers(AsyncLogger.Factory factory) {
      return spyLoggers;
    }
  };

  for (AsyncLogger logger : spyLoggers) {
    futureReturns(GetJournalStateResponseProto.newBuilder()
        .setLastPromisedEpoch(0)
        .setHttpPort(-1)
        .build())
      .when(logger).getJournalState();
    
    futureReturns(
        NewEpochResponseProto.newBuilder().build()
        ).when(logger).newEpoch(Mockito.anyLong());
    
    futureReturns(null).when(logger).format(Mockito.<NamespaceInfo>any());
  }
  
  qjm.recoverUnfinalizedSegments();
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  spyLoggers = ImmutableList.of(
      mockLogger(),
      mockLogger(),
      mockLogger());

  qjm = new QuorumJournalManager(conf, new URI("qjournal://host/jid"),
      FAKE_NSINFO, null, false) {
    @Override
    protected List<AsyncLogger> createLoggers(AsyncLogger.Factory factory) {
      return spyLoggers;
    }
  };

  for (AsyncLogger logger : spyLoggers) {
    GetJournalStateResponseProto p = new GetJournalStateResponseProto();
    p.setLastPromisedEpoch(0);
    p.setHttpPort(-1);
    futureReturns(p)
      .when(logger).getJournalState();
    
    futureReturns(
        new NewEpochResponseProto()
        ).when(logger).newEpoch(Mockito.anyLong());
    
    StartupOption startOpt = null;
    futureReturns(null).when(logger).transitionJournal(
        Mockito.<NamespaceInfo> any(), Mockito.eq(Transition.FORMAT),
        Mockito.eq(startOpt));
  }
  
  qjm.recoverUnfinalizedSegments();
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public ListenableFuture<GetJournalStateResponseProto> getJournalState() {
  return executor.submit(new Callable<GetJournalStateResponseProto>() {
    @Override
    public GetJournalStateResponseProto call() throws IOException {
      GetJournalStateResponseProto ret =
          getProxy().getJournalState(journalIdBytes);
      httpPort = ret.getHttpPort();
      return ret;
    }
  });
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public GetJournalStateResponseProto getJournalState(byte[] journalId)
      throws IOException {
  long epoch = jn.getOrCreateJournal(journalId).getLastPromisedEpoch(); 
  
  GetJournalStateResponseProto ret = new GetJournalStateResponseProto();
  ret.setLastPromisedEpoch(epoch);
  ret.setHttpPort(jn.getBoundHttpAddress().getPort());
  return ret;
}
 

import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos.GetJournalStateResponseProto; //导入依赖的package包/类
@Override
public ListenableFuture<GetJournalStateResponseProto> getJournalState() {
  return executor.submit(new Callable<GetJournalStateResponseProto>() {
    @Override
    public GetJournalStateResponseProto call() throws IOException {
      GetJournalStateResponseProto ret =
          getProxy().getJournalState(journalId);
      httpPort = ret.getHttpPort();
      return ret;
    }
  });
}