KafkaLogAppender extends AppenderSkeleton { public KafkaLogAppender() { } KafkaLogAppender();  KafkaLogAppender(Properties props, String topic); @Override void activateOptions(); @Override void close(); @Override boolean requiresLayout(); String getTopic(); void setTopic(String topic); String getBrokerList(); void setBrokerList(String brokerList); String getSchemaRegistryUrl(); void setSchemaRegistryUrl(String schemaRegistryUrl); static final Long TraceLogTypeId; static final Long DebugLogTypeId; static final Long InfoLogTypeId; static final Long WarnLogTypeId; static final Long ErrorLogTypeId; static final Long FatalLogTypeId; }
@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); }