AbstractFetcherThread.java 文件源码

protected void init() {
    simpleConsumer = new SimpleConsumer(sourceBroker.host, sourceBroker.port, socketTimeout, socketBufferSize, clientId);
    brokerInfo = "host_%s-port_%s".format(sourceBroker.host, sourceBroker.port);
    metricId = new ClientIdAndBroker(clientId, brokerInfo);
    fetcherStats = new FetcherStats(metricId);
    fetcherLagStats = new FetcherLagStats(metricId);
    fetchRequestBuilder = new FetchRequestBuilder().
            clientId(clientId).
            replicaId(fetcherBrokerId).
            maxWait(maxWait).
            minBytes(minBytes);
}