ServiceManagementHandler implements KafkaHandler { @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 (InvalidProtocolBufferException e) { log.error("", e); } } ServiceManagementHandler(ModuleManager moduleManager, KafkaFetcherConfig config); @Override void handle(final ConsumerRecord<String, Bytes> record); @Override String getTopic(); @Override String getConsumePartitions();  }
@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()))); }