try {
GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config);
while (true) {
ConsumerRecords<byte[], byte[]> records = consumer.poll(1000);
for (ConsumerRecord<byte[], byte[]> record : records) {
DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
MetricName.TOTAL_NUMBER);
GenericRecord payload = decoder.decode(record.value());
String clientId = (String) DataUtils.parseRecord(payload, "clientId");
String deviceId = (String) DataUtils.parseRecord(payload, "deviceId");