@Nonnull
    public static <T> Sink<T> remoteReliableTopic(@Nonnull String reliableTopicName,
                                                  @Nonnull ClientConfig clientConfig) {
        String clientXml = asXmlString(clientConfig); //conversion needed for serializability
        return SinkBuilder
                .sinkBuilder("reliableTopicSink(" + reliableTopicName + "))",
                        ctx -> {
                            HazelcastInstance client = newHazelcastClient(asClientConfig(clientXml));
                            ITopic<T> topic = client.getReliableTopic(reliableTopicName);
                            return tuple2(client, topic);
                        })
                .<T>receiveFn((clientTopicTuple, message) -> clientTopicTuple.f1().publish(message))
                .destroyFn(clientTopicTuple -> clientTopicTuple.f0().shutdown())
                .build();
    }
@Test
    public void remoteReliableTopic() {
        // Given
        populateList(srcList);

        List<Object> receivedList = new ArrayList<>();
        remoteHz.getReliableTopic(sinkName).addMessageListener(message -> receivedList.add(message.getMessageObject()));

        // When
        Sink<Object> sink = Sinks.remoteReliableTopic(sinkName, clientConfig);
        p.readFrom(Sources.list(srcName)).writeTo(sink);
        execute();

        // Then
        assertTrueEventually(() -> assertEquals(itemCount, receivedList.size()));
    }