1   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.

1.1   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.

1.1.1   Create write proxy client builder

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

1.1.2   Enable thrift mux

builder = builder.thriftmux(true);

1.1.3   Point the client to write proxy using finagle name

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

1.1.4   Build the write proxy client

DistributedLogClient client = builder.build();

1.2   Create a MultiStreamWriter

1.2.1   Create multi stream writer builder

DistributedLogMultiStreamWriterBuilder builder = DistributedLogMultiStreamWriter.newBuilder();

1.2.2   Build the writer to write a set of streams

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

1.2.4   Configure the flush policy for the multi stream writer

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

1.2.5   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);

1.2.6   Build the multi writer

DistributedLogMultiStreamWriter writer = builder.build();

1.3   Write Records

1.3.1   Write records to multi streams

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

1.3.2   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.
    }
});

1.4   Run the tutorial

Run the example in the following steps:

1.4.1   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

1.4.2   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

1.4.3   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

1.4.4   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

1.4.5   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

1.4.6   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
"""