本文整理了Java中kafka.cluster.Broker类的一些代码示例，展示了Broker类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Broker类的具体详情如下：
包路径：kafka.cluster.Broker
类名称：Broker

Broker介绍

暂无

代码示例

代码示例来源：origin: prestodb/presto

TopicMetadataResponse topicMetadataResponse = simpleConsumer.send(topicMetadataRequest);
    log.debug("Adding Partition %s/%s", metadata.topic(), part.partitionId());
    Broker leader = part.leader();
    if (leader == null) {
      throw new PrestoException(GENERIC_INTERNAL_ERROR, format("Leader election in progress for Kafka topic '%s' partition %s", metadata.topic(), part.partitionId()));
    HostAddress partitionLeader = HostAddress.fromParts(leader.host(), leader.port());

代码示例来源：origin: apache/incubator-gobblin

private void refreshTopicMetadata(KafkaPartition partition) {
 for (String broker : this.brokers) {
  List<TopicMetadata> topicMetadataList = fetchTopicMetadataFromBroker(broker, partition.getTopicName());
  if (topicMetadataList != null && !topicMetadataList.isEmpty()) {
   TopicMetadata topicMetadata = topicMetadataList.get(0);
   for (PartitionMetadata partitionMetadata : topicMetadata.partitionsMetadata()) {
    if (partitionMetadata.partitionId() == partition.getId()) {
     partition.setLeader(partitionMetadata.leader().id(), partitionMetadata.leader().host(), partitionMetadata
       .leader().port());
     break;
    }
   }
   break;
  }
 }
}

代码示例来源：origin: linkedin/kafka-monitor

private List<Broker> brokers(int brokerCount) {
 List<Broker> brokers = new ArrayList<>();
 for (int i = 0; i < brokerCount; i++) {
  brokers.add(new Broker(i, "", -1, null, SecurityProtocol.PLAINTEXT));
 }
 return brokers;
}

代码示例来源：origin: apache/incubator-druid

private boolean isValidNewLeader(Broker broker)
{
 // broker is considered valid new leader if it is not the same as old leaderBroker
 return !(leaderBroker.host().equalsIgnoreCase(broker.host()) && leaderBroker.port() == broker.port());
}

代码示例来源：origin: apache/flink

/**
 * Turn a broker instance into a node instance.
 *
 * @param broker broker instance
 * @return Node representing the given broker
 */
private static Node brokerToNode(Broker broker) {
  return new Node(broker.id(), broker.host(), broker.port());
}

代码示例来源：origin: apache/incubator-druid

private void ensureConsumer(Broker leader) throws InterruptedException
{
 if (consumer == null) {
  while (leaderBroker == null) {
   leaderBroker = findNewLeader(leader);
  }
  log.info(
    "making SimpleConsumer[%s][%s], leader broker[%s:%s]",
    topic, partitionId, leaderBroker.host(), leaderBroker.port()
  );
  consumer = new SimpleConsumer(
    leaderBroker.host(), leaderBroker.port(), SO_TIMEOUT_MILLIS, BUFFER_SIZE, clientId
  );
 }
}

代码示例来源：origin: apache/eagle

public Map<Integer, Long> fetch(String topic, int partitionCount) {
  Map<Integer, PartitionMetadata> metadatas = fetchPartitionMetadata(brokerList, port, topic, partitionCount);
  Map<Integer, Long> ret = new HashMap<>();
  for (int partition = 0; partition < partitionCount; partition++) {
    PartitionMetadata metadata = metadatas.get(partition);
    if (metadata == null || metadata.leader() == null) {
      ret.put(partition, -1L);
      //throw new RuntimeException("Can't find Leader for Topic and Partition. Exiting");
    }
    String leadBroker = metadata.leader().host();
    String clientName = "Client_" + topic + "_" + partition;
    SimpleConsumer consumer = new SimpleConsumer(leadBroker, port, 100000, 64 * 1024, clientName);
    long latestOffset = getLatestOffset(consumer, topic, partition, clientName);
    if (consumer != null) consumer.close();
    ret.put(partition, latestOffset);
  }
  return ret;
}

代码示例来源：origin: com.github.hackerwin7/jlib-utils

SimpleConsumer consumer = null;
try {
  consumer = new SimpleConsumer(seed, port, CONSUME_TIME_OUT, CONSUME_BUFFER_SIZE, clientName);
  List<String> topics = Collections.singletonList(topic);
  TopicMetadataRequest request = new TopicMetadataRequest(topics);
  TopicMetadataResponse response = consumer.send(request);
  List<TopicMetadata> metadatas = response.topicsMetadata();
  for(TopicMetadata item : metadatas) {
    for(PartitionMetadata part : item.partitionsMetadata()) {
      if(part.partitionId() == partition) {
        metadata = part;
        break loop;
} finally {
  if(consumer != null)
    consumer.close();
for(kafka.cluster.Broker replica : metadata.replicas()) {
  replicaBrokers.add(replica.host() + ":" + replica.port());//add replica broker to save

代码示例来源：origin: org.apache.apex/malhar-contrib

private void initializeLastProcessingOffset()
{
 // read last received kafka message
 TopicMetadata tm = KafkaMetadataUtil.getTopicMetadata(Sets.newHashSet((String)getConfigProperties().get(KafkaMetadataUtil.PRODUCER_PROP_BROKERLIST)), this.getTopic());
 if (tm == null) {
  throw new RuntimeException("Failed to retrieve topic metadata");
 }
 partitionNum = tm.partitionsMetadata().size();
 lastMsgs = new HashMap<Integer, Pair<byte[],byte[]>>(partitionNum);
 for (PartitionMetadata pm : tm.partitionsMetadata()) {
  String leadBroker = pm.leader().host();
  int port = pm.leader().port();
  String clientName = this.getClass().getName().replace('$', '.') + "_Client_" + tm.topic() + "_" + pm.partitionId();
  SimpleConsumer consumer = new SimpleConsumer(leadBroker, port, 100000, 64 * 1024, clientName);
  long readOffset = KafkaMetadataUtil.getLastOffset(consumer, tm.topic(), pm.partitionId(), kafka.api.OffsetRequest.LatestTime(), clientName);
  FetchRequest req = new FetchRequestBuilder().clientId(clientName).addFetch(tm.topic(), pm.partitionId(), readOffset - 1, 100000).build();
  FetchResponse fetchResponse = consumer.fetch(req);
  for (MessageAndOffset messageAndOffset : fetchResponse.messageSet(tm.topic(), pm.partitionId())) {
   Message m = messageAndOffset.message();
   ByteBuffer payload = m.payload();
   ByteBuffer key = m.key();
   byte[] valueBytes = new byte[payload.limit()];
   byte[] keyBytes = new byte[key.limit()];
   payload.get(valueBytes);
   key.get(keyBytes);
   lastMsgs.put(pm.partitionId(), new Pair<byte[], byte[]>(keyBytes, valueBytes));
  }
 }
}

代码示例来源：origin: alibaba/jstorm

private SimpleConsumer findLeaderConsumer(int partition) {
  try {
    if (consumer != null) {
      return consumer;
    }
    PartitionMetadata metadata = findLeader(partition);
    if (metadata == null) {
      leaderBroker = null;
      consumer = null;
      return null;
    }
    leaderBroker = metadata.leader();
    consumer = new SimpleConsumer(leaderBroker.host(), leaderBroker.port(), config.socketTimeoutMs, config.socketReceiveBufferBytes,
        config.clientId);
    return consumer;
  } catch (Exception e) {
    LOG.error(e.getMessage(), e);
  }
  return null;
}

代码示例来源：origin: apache/crunch

final SimpleConsumer consumer = getSimpleConsumer(broker);
try {
 topicMetadataResponse = consumer.send(topicMetadataRequest);
 break;
} catch (Exception err) {
 EndPoint endpoint = JavaConversions.seqAsJavaList(broker.endPoints()).get(0);
 LOG.warn(String.format("Fetching topic metadata for topic(s) '%s' from broker '%s' failed",
   Arrays.toString(topics), endpoint.host()), err);
} finally {
 consumer.close();
   new HashMap<>();
 BrokerEndPoint brokerEndPoint = partition.leader();
 if(brokerEndPoint == null){
  throw new CrunchRuntimeException("Unable to find leader for topic:"+metadata.topic()
   +" partition:"+partition.partitionId());
   ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT), SecurityProtocol.PLAINTEXT);
 Broker leader = new Broker(0, JavaConversions.asScalaBuffer(Arrays.asList(endPoint)),
   Option.<String>empty());
  requestInfo = brokerRequests.get(leader);
 requestInfo.put(new TopicAndPartition(metadata.topic(), partition.partitionId()), new PartitionOffsetRequestInfo(
   time, 1));

代码示例来源：origin: linkedin/camus

NUM_TRIES_PARTITION_METADATA);
if (partitionMetadata.errorCode() == ErrorMapping.LeaderNotAvailableCode()) {
 log.info("Skipping the creation of ETL request for Topic : " + topicMetadata.topic()
   + " and Partition : " + partitionMetadata.partitionId() + " Exception : "
   + ErrorMapping.exceptionFor(partitionMetadata.errorCode()));
 reportJobFailureDueToLeaderNotAvailable = true;
} else {
   new LeaderInfo(new URI("tcp://" + partitionMetadata.leader().getConnectionString()),
     partitionMetadata.leader().id());
 if (offsetRequestInfo.containsKey(leader)) {
  ArrayList<TopicAndPartition> topicAndPartitions = offsetRequestInfo.get(leader);

代码示例来源：origin: alibaba/jstorm

fetchResponse = simpleConsumer.fetch(req);
} catch (Exception e) {
  if (e instanceof ConnectException || e instanceof SocketTimeoutException || e instanceof IOException
    LOG.warn("Network error when fetching messages:", e);
    if (simpleConsumer != null) {
      String host = simpleConsumer.host();
      int port = simpleConsumer.port();
      simpleConsumer = null;
      throw new KafkaException("Network error when fetching messages: " + host + ":" + port + " , " + e.getMessage(), e);
    LOG.error("fetch data from kafka topic[" + config.topic + "] host[" + leaderBroker.host() + ":" + leaderBroker.port() + "] partition["
      + partition + "] error:" + code);
  }else {

代码示例来源：origin: couchbase/couchbase-kafka-connector

private String findNewLeader(String oldLeader, String topic, int partition, int port) {
  for (int i = 0; i < 3; i++) {
    PartitionMetadata metadata = findLeader(replicaBrokers, port, topic, partition);
    if (metadata == null
        || metadata.leader() == null
        || oldLeader.equalsIgnoreCase(metadata.leader().host()) && i == 0) {
      // first time through if the leader hasn't changed give ZooKeeper a second to recover
      // second time, assume the broker did recover before failover, or it was a non-Broker issue
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        LOGGER.error("Unable to sleep", e);
      }
    } else {
      return metadata.leader().host();
    }
  }
  throw new IllegalStateException("Unable to find new leader after Broker failure");
}

代码示例来源：origin: apache/incubator-druid

if (metadata != null) {
 replicaBrokers.clear();
 for (Broker replica : metadata.replicas()) {
  replicaBrokers.add(
    HostAndPort.fromParts(replica.host(), replica.port())
  );
 Broker newLeader = metadata.leader();
 if (newLeader != null) {

代码示例来源：origin: Microsoft/Availability-Monitor-for-Kafka

for (kafka.cluster.Broker replica : part.replicas())
  replicas += " " + replica.host();
for (kafka.cluster.Broker replica : part.isr())
  isr += " " + replica.host();
if (part.leader() != null)
  if (part.leader().host() != null)
    leader = part.leader().host();
m_logger.info("    Partition: " + part.partitionId() + ": Leader: " + leader + " Replicas:[" + replicas + "] ISR:[" + isr + "]");

代码示例来源：origin: HomeAdvisor/Kafdrop

private TopicPartitionVO parsePartitionMetadata(String topic, PartitionMetadata pmd)
{
 TopicPartitionVO partition = new TopicPartitionVO(pmd.partitionId());
 if (pmd.leader() != null)
 {
   partition.addReplica(new TopicPartitionVO.PartitionReplica(pmd.leader().id(), true, true));
 }
 final List<Integer> isr = getIsr(topic, pmd);
 pmd.replicas().stream()
   .map(replica -> new TopicPartitionVO.PartitionReplica(replica.id(), isr.contains(replica.id()), false))
   .forEach(partition::addReplica);
 return partition;
}

代码示例来源：origin: linkedin/camus

private TopicMetadataResponse mockTopicMetaDataResponse() {
 PartitionMetadata pMeta = EasyMock.createMock(PartitionMetadata.class);
 mocks.add(pMeta);
 EasyMock.expect(pMeta.errorCode()).andReturn((short)0).anyTimes();
 Broker broker = new Broker(0, "localhost", 2121);
 EasyMock.expect(pMeta.leader()).andReturn(broker).anyTimes();
 EasyMock.expect(pMeta.partitionId()).andReturn(PARTITION_1_ID).anyTimes();
 List<PartitionMetadata> partitionMetadatas = new ArrayList<PartitionMetadata>();
 partitionMetadatas.add(pMeta);    
 TopicMetadata tMeta = EasyMock.createMock(TopicMetadata.class);
 mocks.add(tMeta);
 EasyMock.expect(tMeta.topic()).andReturn(TOPIC_1).anyTimes();
 EasyMock.expect(tMeta.errorCode()).andReturn((short)0).anyTimes();
 EasyMock.expect(tMeta.partitionsMetadata()).andReturn(partitionMetadatas).anyTimes();
 List<TopicMetadata> topicMetadatas = new ArrayList<TopicMetadata>();
 topicMetadatas.add(tMeta);
 TopicMetadataResponse metadataResponse = EasyMock.createMock(TopicMetadataResponse.class);
 mocks.add(metadataResponse);
 EasyMock.expect(metadataResponse.topicsMetadata()).andReturn(topicMetadatas).anyTimes();
 return metadataResponse;
}

代码示例来源：origin: apache/incubator-druid

private long getOffset(boolean earliest) throws InterruptedException
{
 TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partitionId);
 Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
 requestInfo.put(
   topicAndPartition,
   new PartitionOffsetRequestInfo(
     earliest ? kafka.api.OffsetRequest.EarliestTime() : kafka.api.OffsetRequest.LatestTime(), 1
   )
 );
 OffsetRequest request = new OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientId);
 OffsetResponse response;
 try {
  response = consumer.getOffsetsBefore(request);
 }
 catch (Exception e) {
  ensureNotInterrupted(e);
  log.error(e, "caught exception in getOffsetsBefore [%s] - [%s]", topic, partitionId);
  return -1;
 }
 if (response.hasError()) {
  log.error(
    "error fetching data Offset from the Broker [%s]. reason: [%s]", leaderBroker.host(),
    response.errorCode(topic, partitionId)
  );
  return -1;
 }
 long[] offsets = response.offsets(topic, partitionId);
 return earliest ? offsets[0] : offsets[offsets.length - 1];
}

代码示例来源：origin: linkedin/kafka-monitor

static boolean someBrokerNotPreferredLeader(List<PartitionInfo> partitionInfoList, Collection<Broker> brokers) {
 Set<Integer> brokersNotPreferredLeader = new HashSet<>(brokers.size());
 for (Broker broker: brokers)
  brokersNotPreferredLeader.add(broker.id());
 for (PartitionInfo partitionInfo : partitionInfoList)
  brokersNotPreferredLeader.remove(partitionInfo.replicas()[0].id());
 return !brokersNotPreferredLeader.isEmpty();
}