website/docs/latest/tutorials/basic-3.rst

---
title: API - Write Records to Multiple Streams
layout: default
---

.. contents:: Basic Tutorial - Write Records to Multiple Streams

Write Records to Multiple Streams
=================================

This tutorial shows how to write records using write proxy multi stream writer. The `DistributedLogMultiStreamWriter`
is a wrapper over `DistributedLogClient` on writing records to a set of streams in a `round-robin` way and ensure low write latency even on single stream ownership failover.

.. sectnum::

Open a write proxy client
~~~~~~~~~~~~~~~~~~~~~~~~~

Before everything, you have to open a write proxy client to write records.
These are the steps to follow to `open a write proxy client`.

Create write proxy client builder
---------------------------------

::

        DistributedLogClientBuilder builder = DistributedLogClientBuilder.newBuilder()
                .clientId(ClientId.apply("console-proxy-writer"))
                .name("console-proxy-writer");


Enable thrift mux
-----------------

::

        builder = builder.thriftmux(true);


Point the client to write proxy using finagle name
--------------------------------------------------

::

        String finagleNameStr = "inet!127.0.0.1:8000";
        builder = builder.finagleNameStr(finagleNameStr);


Build the write proxy client
----------------------------

::

        DistributedLogClient client = builder.build();


Create a `MultiStreamWriter`
~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Create multi stream writer builder
----------------------------------

::

        DistributedLogMultiStreamWriterBuilder builder = DistributedLogMultiStreamWriter.newBuilder();


Build the writer to write a set of streams
------------------------------------------

::

        List<String> streams = ...;
        builder = builder.streams(streams);


Point the multi stream writer to use write proxy client
-------------------------------------------------------

::

        builder = builder.client(client);


Configure the flush policy for the multi stream writer
------------------------------------------------------

::

        // transmit immediately after a record is written.
        builder = builder.bufferSize(0);
        builder = builder.flushIntervalMs(0);


Configure the request timeouts and retry policy for the multi stream writer
---------------------------------------------------------------------------

::

        // Configure the speculative timeouts - if writing to a stream cannot
        // complete within the speculative timeout, it would try writing to
        // another streams.
        builder = builder.firstSpeculativeTimeoutMs(10000)
        builder = builder.maxSpeculativeTimeoutMs(20000)
        // Configure the request timeout.
        builder = builder.requestTimeoutMs(50000);


Build the multi writer
----------------------

::

        DistributedLogMultiStreamWriter writer = builder.build();


Write Records
~~~~~~~~~~~~~

Write records to multi streams 
------------------------------

::

        byte[] data = ...;
        Future<DLSN> writeFuture = writer.write(ByteBuffer.wrap(data));


Register the write callback
---------------------------

Register a future listener on write completion.

::

        writeFuture.addEventListener(new FutureEventListener<DLSN>() {
            @Override
            public void onFailure(Throwable cause) {
                // executed when write failed.
            }

            @Override
            public void onSuccess(DLSN value) {
                // executed when write completed.
            }
        });


Run the tutorial
~~~~~~~~~~~~~~~~

Run the example in the following steps:

Start the local bookkeeper cluster
----------------------------------

You can use follow command to start the distributedlog stack locally.
After the distributedlog is started, you could access it using
distributedlog uri *distributedlog://127.0.0.1:7000/messaging/distributedlog*.

::

        // dlog local ${zk-port}
        ./distributedlog-core/bin/dlog local 7000


Start the write proxy
---------------------

Start the write proxy, listening on port 8000.

::

        // DistributedLogServerApp -p ${service-port} --shard-id ${shard-id} -sp ${stats-port} -u {distributedlog-uri} -mx -c ${conf-file}
        ./distributedlog-proxy-server/bin/dlog org.apache.distributedlog.service.DistributedLogServerApp -p 8000 --shard-id 1 -sp 8001 -u distributedlog://127.0.0.1:7000/messaging/distributedlog -mx -c ${distributedlog-repo}/distributedlog-proxy-server/conf/distributedlog_proxy.conf


Create multiple streams
-----------------------

Create multiple streams under the distributedlog uri.

::

        // Create Stream `basic-stream-{3-7}`
        // dlog tool create -u ${distributedlog-uri} -r ${stream-prefix} -e ${stream-regex}
        ./distributedlog-core/bin/dlog tool create -u distributedlog://127.0.0.1:7000/messaging/distributedlog -r basic-stream- -e 3-7


Tail the streams
----------------

Tailing the streams using `MultiReader` to wait for new records.

::

        // Tailing Stream `basic-stream-{3-7}`
        // runner run org.apache.distributedlog.basic.MultiReader ${distributedlog-uri} ${stream}[,${stream}]
        ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.MultiReader distributedlog://127.0.0.1:7000/messaging/distributedlog basic-stream-3,basic-stream-4,basic-stream-5,basic-stream-6,basic-stream-7


Write the records
-----------------

Run the example to write records to the multi streams in a console.

::

        // Write Records into Stream `basic-stream-{3-7}`
        // runner run org.apache.distributedlog.basic.ConsoleProxyMultiWriter ${distributedlog-uri} ${stream}[,${stream}]
        ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.ConsoleProxyMultiWriter 'inet!127.0.0.1:8000' basic-stream-3,basic-stream-4,basic-stream-5,basic-stream-6,basic-stream-7

Check the results
-----------------

Example output from `ConsoleProxyMultiWriter` and `MultiReader`.

::

        // Output of `ConsoleProxyWriter`
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[inet] = com.twitter.finagle.InetResolver(com.twitter.finagle.InetResolver@fbb628c)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[fixedinet] = com.twitter.finagle.FixedInetResolver(com.twitter.finagle.FixedInetResolver@5a25adb1)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[neg] = com.twitter.finagle.NegResolver$(com.twitter.finagle.NegResolver$@5fae6db3)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[nil] = com.twitter.finagle.NilResolver$(com.twitter.finagle.NilResolver$@34a433d8)
        May 08, 2016 11:09:21 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply
        INFO: Resolver[fail] = com.twitter.finagle.FailResolver$(com.twitter.finagle.FailResolver$@847c4e8)
        May 08, 2016 11:09:22 AM com.twitter.finagle.Init$$anonfun$1 apply$mcV$sp
        [dlog] > message-1
        [dlog] > message-2
        [dlog] > message-3
        [dlog] > message-4
        [dlog] > message-5
        [dlog] >


        // Output of `MultiReader`
        Opening log stream basic-stream-3
        Opening log stream basic-stream-4
        Opening log stream basic-stream-5
        Opening log stream basic-stream-6
        Opening log stream basic-stream-7
        Log stream basic-stream-4 is empty.
        Wait for records from basic-stream-4 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-4
        Log stream basic-stream-5 is empty.
        Wait for records from basic-stream-5 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-5
        Log stream basic-stream-6 is empty.
        Wait for records from basic-stream-6 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-6
        Log stream basic-stream-3 is empty.
        Wait for records from basic-stream-3 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-3
        Log stream basic-stream-7 is empty.
        Wait for records from basic-stream-7 starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0}
        Open reader to read records from stream basic-stream-7
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-4
        """
        message-1
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-6
        """
        message-2
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-3
        """
        message-3
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-7
        """
        message-4
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} from stream basic-stream-5
        """
        message-5
        """