@Override
    public void handle(final ConsumerRecord<String, Bytes> record) {
        try {
            if (record.key().startsWith("register-")) {
                serviceReportProperties(InstanceProperties.parseFrom(record.value().get()));
            } else {
                keepAlive(InstancePingPkg.parseFrom(record.value().get()));
            }
        } catch (Exception e) {
            log.error("handle record failed", e);
        }
    }
@Test
    public void testHandler() {
        InstanceProperties properties = InstanceProperties.newBuilder()
                                                          .setService(SERVICE)
                                                          .setServiceInstance(SERVICE_INSTANCE)
                                                          .build();
        InstancePingPkg ping = InstancePingPkg.newBuilder()
                                              .setService(SERVICE)
                                              .setServiceInstance(SERVICE_INSTANCE)
                                              .build();

        handler.handle(new ConsumerRecord<>(TOPIC_NAME, 0, 0, "register", Bytes.wrap(properties.toByteArray())));
        handler.handle(
            new ConsumerRecord<>(TOPIC_NAME, 0, 0, ping.getServiceInstance(), Bytes.wrap(ping.toByteArray())));
    }