import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
@Test
public void testKafkaLogAppender() {
    Properties consumerProps = new Properties();
    consumerProps.put("zookeeper.connect", zookeeper);
    consumerProps.put("group.id", "kafka-log-appender-test");
    consumerProps.put("auto.offset.reset", "smallest");
    consumerProps.put("schema.registry.url", schemaRegistry);

    Map<String, Integer> topicMap = new HashMap<String, Integer>();
    topicMap.put(topic, 1);

    ConsumerIterator<String, Object> iterator = Consumer.createJavaConsumerConnector(new ConsumerConfig(consumerProps))
            .createMessageStreams(topicMap, new StringDecoder(null), new KafkaAvroDecoder(new VerifiableProperties(consumerProps)))
            .get(topic).get(0).iterator();

    String testMessage = "I am a test message";
    logger.info(testMessage);

    MessageAndMetadata<String, Object> messageAndMetadata = iterator.next();
    GenericRecord logLine = (GenericRecord) messageAndMetadata.message();
    assertEquals(logLine.get("line").toString(), testMessage);
    assertEquals(logLine.get("logtypeid"), KafkaLogAppender.InfoLogTypeId);
    assertNotNull(logLine.get("source"));
    assertEquals(((Map<CharSequence, Object>) logLine.get("timings")).size(), 1);
    assertEquals(((Map<CharSequence, Object>) logLine.get("tag")).size(), 2);
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
@Override
public String deserialize(byte[] message) {
    if (kafkaAvroDecoder == null) {
        SchemaRegistryClient schemaRegistry = new CachedSchemaRegistryClient(this.schemaRegistryUrl, this.identityMapCapacity);
        this.kafkaAvroDecoder = new KafkaAvroDecoder(schemaRegistry);
    }
    return this.kafkaAvroDecoder.fromBytes(message).toString();
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
private static void processStream(JavaStreamingContext ssc, JavaSparkContext sc) {
  System.out.println("--> Processing stream");

  Map<String, String> props = new HashMap<>();
  props.put("bootstrap.servers", "localhost:9092");
  props.put("schema.registry.url", "http://localhost:8081");
  props.put("group.id", "spark");
  props.put("specific.avro.reader", "true");

  props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

  Set<String> topicsSet = new HashSet<>(Collections.singletonList("test"));

  JavaPairInputDStream<String, Object> stream = KafkaUtils.createDirectStream(ssc, String.class, Object.class,
    StringDecoder.class, KafkaAvroDecoder.class, props, topicsSet);

  stream.foreachRDD(rdd -> {
    rdd.foreachPartition(iterator -> {
        while (iterator.hasNext()) {
          Tuple2<String, Object> next = iterator.next();
          Model model = (Model) next._2();
          System.out.println(next._1() + " --> " + model);
        }
      }
    );
  });
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
@Test(dataProvider = "kafkaMetrics")
public void schemaRegistryKafkaReporterTest(HashMap<String,String> tags, String metricKey, Double value,
    Double truncatedValue,
    long timestamp) {

  /* GIVEN: A Zookeeper instance, a Kafka broker, and a the Schema Registry-based Kafka reporter we're testing */
  initializeSchemaRegistryReporter();
  SimpleConsumer kafkaConsumer = new SimpleConsumer(testKafkaHost, testKafkaPort, 10000, 1024000, "simpleConsumer");
  KafkaAvroDecoder decoder = new KafkaAvroDecoder(schemaRegistryClient);

  /* WHEN: A new metric is appended to the reporter's buffer and we tell the reporter to send its data */
  submitMetricToReporter(tags, metricKey, value, timestamp);

  /* WHEN: A Kafka consumer reads the latest message from the same topic on the Kafka server */
  byte[] bytes = fetchLatestRecordPayloadBytes(kafkaConsumer);
  
  /* WHEN: The latest message is decoded using the Schema Regsitry based decoder */
  GenericRecord result = null;
  try {
    result = (GenericRecord) decoder.fromBytes(bytes);
  }
  catch (SerializationException e) {
    fail("Failed to deserialize message:" + e.getMessage());
  }

  /* THEN: The field values of the decoded record should be the same as those of the input fields. */
  assertThat(result).isNotNull();
  assertThat(result.get("prefix")).isEqualTo(TagsHelper.constructMetricPrefix(TagsHelper.DEFAULT_PREFIX, tags));
  assertThat(result.get("reportTime")).isEqualTo(timestamp);
  assertThat(((Map) result.get("metricValues")).get(metricKey)).isEqualTo(truncatedValue);
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
@Test
    public void testCodahaleKafkaMetricsReporter() {
        registry = new MetricRegistry();
        registry.counter("test_counter").inc();

        kafkaReporter = KafkaReporter.builder(registry,
                kafkaConnect,
                topic,
                schemaRegistry).build();

//        ObjectMapper mapper = new ObjectMapper().registerModule(new MetricsModule(TimeUnit.SECONDS,
//                TimeUnit.SECONDS,
//                false));
//        StringWriter r = new StringWriter();
//        try {
//            mapper.writeValue(r, registry);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }

        kafkaReporter.report();

        Properties props = new Properties();
        props.put("zookeeper.connect", zkConnect);
        props.put("group.id", UUID.randomUUID().toString());
        props.put("auto.offset.reset", "smallest");
        props.put("zookeeper.session.timeout.ms", "30000");
        props.put("consumer.timeout.ms", "30000");
        props.put("schema.registry.url", schemaRegistry);

        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
        KafkaStream<String, Object> messageStream = consumer.createMessageStreamsByFilter(new Whitelist(topic),
                1,
                new StringDecoder(null),
                new KafkaAvroDecoder(new VerifiableProperties(props))).get(0);

        GenericRecord message = (GenericRecord) messageStream.iterator().next().message();
        assertNotNull(message);
    }
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
@Test
public void testTopicReporter() {
    MetricsRegistry registry = new MetricsRegistry();
    Counter counter = registry.newCounter(KafkaReporterTest.class, "test-counter");
    counter.inc();

    Properties producerProps = new Properties();
    producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnect);
    producerProps.put("schema.registry.url", schemaRegistry);

    KafkaReporter reporter = new KafkaReporter(registry, producerProps, topic);
    reporter.start(1, TimeUnit.SECONDS);

    Properties props = new Properties();
    props.put("zookeeper.connect", zkConnect);
    props.put("group.id", UUID.randomUUID().toString());
    props.put("auto.offset.reset", "smallest");
    props.put("zookeeper.session.timeout.ms", "30000");
    props.put("consumer.timeout.ms", "30000");
    props.put("schema.registry.url", schemaRegistry);

    ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
    KafkaStream<String, Object> messageStream = consumer.createMessageStreamsByFilter(new Whitelist(topic),
            1,
            new StringDecoder(null),
            new KafkaAvroDecoder(new VerifiableProperties(props))).get(0);

    GenericRecord message = (GenericRecord) messageStream.iterator().next().message();
    assertNotNull(message);

    reporter.shutdown();
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
private void initKafka() {
    schemaRegistryClient = new MockSchemaRegistryClient();
    kafkaAvroDecoder = new KafkaAvroDecoder(schemaRegistryClient);
    Properties defaultConfig = new Properties();
    defaultConfig.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "bogus");
    avroSerializer = new KafkaAvroSerializer(schemaRegistryClient);
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
public AvroSerde(VerifiableProperties encoderProps, VerifiableProperties decoderProps) {
    encoder = new KafkaAvroEncoder(encoderProps);
    decoder = new KafkaAvroDecoder(decoderProps);
}
 

import io.confluent.kafka.serializers.KafkaAvroDecoder; //导入依赖的package包/类
protected void init(SecorConfig config) {
    decoder = new KafkaAvroDecoder(schemaRegistryClient);
}