中国民生银行大数据团队Kafka1.X管控实践

  DescribeClusterOptions describeClusterOptions =
      new DescribeClusterOptions().timeoutMs((int) kafkaAdminClientGetTimeoutMs);

  DescribeClusterResult describeClusterResult =
      kafkaAdminClient.describeCluster(describeClusterOptions);

  KafkaFuture<String> clusterIdFuture = describeClusterResult.clusterId();
  KafkaFuture<Node> controllerFuture = describeClusterResult.controller();
  KafkaFuture<Collection<Node>> nodesFuture = describeClusterResult.nodes();
  String clusterId = "";
  Node controller = null;
  Collection<Node> nodes = new ArrayList<>();

  try {
    clusterId = clusterIdFuture.get(kafkaAdminClientGetTimeoutMs, TimeUnit.MILLISECONDS);
    controller = controllerFuture.get(kafkaAdminClientGetTimeoutMs, TimeUnit.MILLISECONDS);
    nodes = nodesFuture.get(kafkaAdminClientGetTimeoutMs, TimeUnit.MILLISECONDS);
  } catch (Exception exception) {
    log.warn("Describe cluster exception:" + exception);
    throw new ApiException("Describe cluster exception:" + exception);
  } finally {
    if (clusterIdFuture.isDone() && !clusterIdFuture.isCompletedExceptionally()) {
      //        clusterDetail.put("clusterId", clusterId);
      clusterInfo.setClusterId(clusterId);
    }
    if (controllerFuture.isDone() && !controllerFuture.isCompletedExceptionally()) {
      //        clusterDetail.put("controllerId", controller);
      clusterInfo.setController(controller);
    }
    if (nodesFuture.isDone() && !nodesFuture.isCompletedExceptionally()) {
      //        clusterDetail.put("nodes", nodes);
      clusterInfo.setNodes(nodes);
    }
  }

  String brokerInfoStr = null;
  try {
    // TODO replace zkClient with kafkaZKClient
    brokerInfoStr =
        new String(
            zkClient
                .getData()
                .forPath(ZkUtils.BrokerIdsPath() + "/" + entry.getKey()));
  } catch (Exception e) {
    e.printStackTrace();
  }
  BrokerInfo brokerInfo;
  try {
    ObjectMapper objectMapper = new ObjectMapper();
    brokerInfo = objectMapper.readValue(brokerInfoStr, BrokerInfo.class);
  } catch (Exception exception) {
    throw new ApiException("List broker exception." + exception);
}

    * Version 3 JSON schema for a broker is:
    * {
    *   "version":3,
    *   "host":"localhost",
    *   "port":9092,
    *   "jmx_port":9999,
    *   "timestamp":"2233345666",
    *   "endpoints":["PLAINTEXT://host1:9092", "SSL://host1:9093"],
    *   "rack":"dc1"
    * }
    *
    * Version 4 (current) JSON schema for a broker is:
    * {
    *   "version":4,
    *   "host":"localhost",
    *   "port":9092,
    *   "jmx_port":9999,
    *   "timestamp":"2233345666",
    *   "endpoints":["CLIENT://host1:9092", "REPLICATION://host1:9093"],
    *   "listener_security_protocol_map":{"CLIENT":"SSL", "REPLICATION":"PLAINTEXT"},
    *   "rack":"dc1"
    * }

  // Return <Current partition replica assignment, Proposed partition reassignment>
  public List<ReassignModel> generateReassignPartition(ReassignWrapper reassignWrapper) {
    KafkaZkClient kafkaZkClient = zookeeperUtils.getKafkaZkClient();
    List<ReassignModel> result = new ArrayList<>();

    Seq brokerSeq =
        JavaConverters.asScalaBufferConverter(reassignWrapper.getBrokers()).asScala().toSeq();
    // <Proposed partition reassignment，Current partition replica assignment>
    Tuple2 resultTuple2;
    try {
      resultTuple2 =
          ReassignPartitionsCommand.generateAssignment(
              kafkaZkClient, brokerSeq, reassignWrapper.generateReassignJsonString(), false);
    } catch (Exception exception) {
      throw new ApiException("Generate reassign plan exception." + exception);
    }
    HashMap<TopicPartitionReplica, String> emptyMap = new HashMap<>();
    ObjectMapper objectMapper = new ObjectMapper();
    try {
      result.add(
          objectMapper.readValue(
              ReassignPartitionsCommand.formatAsReassignmentJson(
                  (scala.collection.Map<TopicPartition, Seq<Object>>) resultTuple2._2(),
                  JavaConverters.mapAsScalaMapConverter(emptyMap).asScala()),
              ReassignModel.class));
      result.add(
          objectMapper.readValue(
              ReassignPartitionsCommand.formatAsReassignmentJson(
                  (scala.collection.Map<TopicPartition, Seq<Object>>) resultTuple2._1(),
                  JavaConverters.mapAsScalaMapConverter(emptyMap).asScala()),
              ReassignModel.class));
      Collections.sort(result.get(0).getPartitions());
      Collections.sort(result.get(1).getPartitions());
    } catch (Exception exception) {
      throw new ApiException("Generate reassign plan exception." + exception);
    }

    return result;
  }

  private Object avroDeserialize(byte[] bytes, String avroSchema, boolean isInSchemaRegistry) {
    Schema schema = new Schema.Parser().parse(avroSchema);
    DatumReader reader = new GenericDatumReader<GenericRecord>(schema);
    ByteBuffer buffer = ByteBuffer.wrap(bytes);
    Object object = null;

    if (isInSchemaRegistry) {
      try {
        //这里利用confluent提供的io.confluent.kafka.serializers.KafkaAvroDeserializer进行解码
        object = confluentSchemaService.deserializeBytesToObject("", bytes, schema);
      } catch (SerializationException serializationException) {
        throw new ApiException("Avro Deserialize exception. " + serializationException);
      }
    } else {
      try {
        object =
            reader.read(
                null,
                DecoderFactory.get().binaryDecoder(buffer.array(), 0, bytes.length, null));
      } catch (IOException exception) {
        throw new ApiException("Avro Deserialize exception. " + exception);
      }
    }

    return object;
  }

    ConsumerRecords<Object, Object> crs = consumer.poll(timeoutMs);
    if (crs.count() != 0) {
      Iterator<ConsumerRecord<Object, Object>> it = crs.iterator();
      while (it.hasNext()) {
        Record record = Record.builder().topic(topic).keyDecoder(keyDecoder)
            .valueDecoder(valueDecoder).build();
        ConsumerRecord<Object, Object> initCr = it.next();
        record.setOffset(initCr.offset());
        record.setTimestamp(initCr.timestamp());
        record.setKey(initCr.key());
        record.setValue(initCr.value());
        recordList.add(record);
      }
    }

    public String getValueByDecoder(String decoder, Object value) {
    if (value == null) return null;
    Class<?> type = KafkaUtils.DESERIALIZER_TYPE_MAP.get(decoder);
    try {
      if (String.class.isAssignableFrom(type)) {
        return value.toString();
      }

      if (Short.class.isAssignableFrom(type)) {
        return value.toString();
      }
      ...

      if (Bytes.class.isAssignableFrom(type)) {
        Bytes bytes = (Bytes) value;
        return bytes.toString();
      }

      if (byte[].class.isAssignableFrom(type)) {
        if (decoder.contains("AvroDeserializer")) {
          return value.toString();
        } else {
          byte[] byteArray = (byte[]) value;
          return new String(byteArray);
        }
      }

      if (ByteBuffer.class.isAssignableFrom(type)) {
        ByteBuffer byteBuffer = (ByteBuffer) value;
        return new String(byteBuffer.array());
      }
    }
    ...

public Map<HostAndPort, ZkServerStat> stat() {
List<HostAndPort> hostAndPortList = zookeeperUtils.getZookeeperConfig().getHostAndPort();
Map<HostAndPort, ZkServerStat> result = new HashMap<>();
for (int i = 0; i < hostAndPortList.size(); i++) {
  HostAndPort hp = hostAndPortList.get(i);
  try {
    result.put(
        hp,
        zookeeperUtils.parseStatResult(
            zookeeperUtils.executeCommand(
                hp.getHostText(), hp.getPort(), ZkServerCommand.stat.toString())));
  } catch (ServiceNotAvailableException serviceNotAvailbleException) {
    log.warn(
        "Execute "
            + ZkServerCommand.stat.toString()
            + " command failed. Exception:"
            + serviceNotAvailbleException);
    result.put(
        hp,
        ZkServerStat.builder()
            .mode(serviceNotAvailbleException.getServiceState())
            .msg(serviceNotAvailbleException.getMessage())
            .build());
  }
}
return result;
}

 public List<SchemaRegistryMetadata> getAllSubjects() {
    try {
      Collection<String> subjects = this.schemaRegistryClient.getAllSubjects();
      List<SchemaRegistryMetadata> allSubjects = new ArrayList<>();
      SchemaMetadata schemaMetadata;
      SchemaRegistryMetadata schemaRegistryMetadata;

      for (String subject : subjects) {
        schemaMetadata = schemaRegistryClient.getLatestSchemaMetadata(subject);
        schemaRegistryMetadata = SchemaRegistryMetadata.builder().subject(subject)
            .id(schemaMetadata.getId())
            .version(schemaMetadata.getVersion()).schema(schemaMetadata.getSchema()).build();
        allSubjects.add(schemaRegistryMetadata);
      }
      return allSubjects;
    } catch (Exception exception) {
      throw new ApiException("ConfluentSchemaService getAllSubjects exception : " + exception);
    }
  }