topic-with-data-serializable

    public static void testTopicWithDataSerializable() throws InterruptedException {
        final HazelcastInstance instance1 = Hazelcast.newHazelcastInstance();
        final HazelcastInstance instance2 = Hazelcast.newHazelcastInstance();
        final CountDownLatch latch = new CountDownLatch(1);
        final ITopic<ConfigurationChangeMessage> publisher = instance1.getTopic("message");
        final ITopic<ConfigurationChangeMessage> consumer = instance2.getTopic("message");
        byte[] val = new byte[256];
        final Random random = new Random(System.currentTimeMillis());
        random.nextBytes(val);
        final ConfigurationChangeMessage confMessage = new ConfigurationChangeMessage("type", "name", val);
        consumer.addMessageListener(new MessageListener<ConfigurationChangeMessage>() {
            public void onMessage(Message<ConfigurationChangeMessage> m) {
                final ConfigurationChangeMessage message = m.getMessageObject();
                System.err.println("message : " + message);
                latch.countDown();
            }
        });

        publisher.publish(confMessage);

        assertTrue(latch.await(10, TimeUnit.SECONDS));
    }

    public class ConfigurationChangeMessage implements DataSerializable {

        private String configurationType;
        private String name;
        private byte[] value;

        public ConfigurationChangeMessage() {
        }

        public ConfigurationChangeMessage(String configurationType, String name, byte[] value) {
            this.configurationType = configurationType;
            this.name = name;
            this.value = value;
        }

        public String getConfigurationType() {
            return configurationType;
        }

        public String getName() {
            return name;
        }

        public byte[] getValue() {
            return value;
        }

        public void writeData(ObjectDataOutput objectDataOutput) throws IOException {
            objectDataOutput.writeUTF(configurationType);
            objectDataOutput.writeUTF(name);
            objectDataOutput.writeInt(value.length);
            objectDataOutput.write(value);
        }

        public void readData(ObjectDataInput objectDataInput) throws IOException {
            configurationType = objectDataInput.readUTF();
            name = objectDataInput.readUTF();
            int size = objectDataInput.readInt();
            value = new byte[size];
            objectDataInput.readFully(value);
        }

    }

And a nullsafe version:

    public class ConfigurationChangeMessage implements DataSerializable {

        private String configurationType;
        private String name;
        private byte[] value;

        public ConfigurationChangeMessage() {
        }

        public ConfigurationChangeMessage(String configurationType, String name, byte[] value) {
            this.configurationType = configurationType;
            this.name = name;
            this.value = value;
        }

        public String getConfigurationType() {
            return configurationType;
        }

        public String getName() {
            return name;
        }

        public byte[] getValue() {
            return value;
        }

        public void writeData(ObjectDataOutput objectDataOutput) throws IOException {
            objectDataOutput.writeUTF(configurationType);
            objectDataOutput.writeUTF(name);
            objectDataOutput.writeInt(value == null ? -1 : value.length);
            if (value != null) {
              objectDataOutput.write(value);
            }
        }

        public void readData(ObjectDataInput objectDataInput) throws IOException {
            configurationType = objectDataInput.readUTF();
            name = objectDataInput.readUTF();
            int size = objectDataInput.readInt();
            if (size != -1) {
              value = new byte[size];
              objectDataInput.readFully(value);
            }
        }

    }