### Question:
NettyBroadcastService implements ManagedBroadcastService { @Override public void broadcast(String subject, byte[] payload) { if (enabled) { Message message = new Message(subject, payload); byte[] bytes = SERIALIZER.encode(message); ByteBuf buf = serverChannel.alloc().buffer(4 + bytes.length); buf.writeInt(bytes.length).writeBytes(bytes); serverChannel.writeAndFlush(new DatagramPacket(buf, groupAddress)); } } NettyBroadcastService(Address localAddress, Address groupAddress, boolean enabled); static Builder builder(); @Override void broadcast(String subject, byte[] payload); @Override synchronized void addListener(String subject, Consumer<byte[]> listener); @Override synchronized void removeListener(String subject, Consumer<byte[]> listener); @Override CompletableFuture<BroadcastService> start(); @Override boolean isRunning(); @Override CompletableFuture<Void> stop();  }### Answer:
@Test public void testBroadcast() throws Exception { netty1.addListener("test", bytes -> { threadAssertEquals(0, bytes.length); resume(); }); netty2.broadcast("test", new byte[0]); await(5000); }