Basic Tutorial - Tail reading records from a stream
1 Tail reading records from a stream
This tutorial shows how to tail read records from a stream.
1.1 Open a distributedlog manager
1.1.1 Create distributedlog URI
String dlUriStr = ...; URI uri = URI.create(dlUriStr);
1.1.2 Create distributedlog configuration
DistributedLogConfiguration conf = new DistributedLogConfiguration();
1.1.3 Build the distributedlog namespace
DistributedLogNamespace namespace = DistributedLogNamespaceBuilder.newBuilder() .conf(conf) .uri(uri) .build();
1.1.4 Open the distributedlog manager
DistributedLogManager dlm = namespace.openLog("basic-stream-9");
1.2 Get Last Record
Get the last record from the record. From the last record, we can use DLSN of last record to start tailing the stream.
LogRecordWithDLSN record = dlm.getLastLogRecord(); DLSN lastDLSN = record.getDlsn();
1.3 Read Records
1.3.1 Open the stream
Open the stream to start read the records.
AsyncLogReader reader = FutureUtils.result(dlm.openAsyncLogReader(lastDLSN));
1.3.2 Read the records
Read the next available record from the stream. The future is satisified when the record is available.
Future<LogRecordWithDLSN> readFuture = reader.readNext();
1.3.3 Register the read callback
Register a future listener on read completion. The reader will be notified once the record is ready for consuming.
final FutureEventListener<LogRecordWithDLSN> readListener = new FutureEventListener<LogRecordWithDLSN>() { @Override public void onFailure(Throwable cause) { // executed when read failed. } @Override public void onSuccess(LogRecordWithDLSN record) { // process the record ... // issue read next reader.readNext().addEventListener(this); } }; reader.readNext().addEventListener(readListener);
1.5 Run the tutorial
Run the example in the following steps:
1.5.1 Start the local bookkeeper cluster
You can use follow command to start the distributedlog stack locally. After the distributedlog cluster 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.5.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.5.3 Create the stream
Create the stream under the distributedlog uri.
// Create Stream `basic-stream-9` // 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 9
1.5.4 Tail the stream
Tailing the stream using TailReader to wait for new records.
// Tailing Stream `basic-stream-9` // runner run org.apache.distributedlog.basic.TailReader ${distributedlog-uri} ${stream} ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.TailReader distributedlog://127.0.0.1:7000/messaging/distributedlog basic-stream-9
1.5.5 Write records
Run the example to write records to the stream in a console.
// Write Records into Stream `basic-stream-9` // runner run org.apache.distributedlog.basic.ConsoleProxyWriter ${distributedlog-uri} ${stream} ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.ConsoleProxyWriter 'inet!127.0.0.1:8000' basic-stream-9
1.5.6 Check the results
Example output from ConsoleProxyWriter and TailReader.
// Output of `ConsoleProxyWriter` May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply INFO: Resolver[inet] = com.twitter.finagle.InetResolver(com.twitter.finagle.InetResolver@756d7bba) May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply INFO: Resolver[fixedinet] = com.twitter.finagle.FixedInetResolver(com.twitter.finagle.FixedInetResolver@1d2e91f5) May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply INFO: Resolver[neg] = com.twitter.finagle.NegResolver$(com.twitter.finagle.NegResolver$@5c707aca) May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply INFO: Resolver[nil] = com.twitter.finagle.NilResolver$(com.twitter.finagle.NilResolver$@5c8d932f) May 08, 2016 10:27:41 AM com.twitter.finagle.BaseResolver$$anonfun$resolvers$1 apply INFO: Resolver[fail] = com.twitter.finagle.FailResolver$(com.twitter.finagle.FailResolver$@52ba2221) May 08, 2016 10:27:41 AM com.twitter.finagle.Init$$anonfun$1 apply$mcV$sp [dlog] > test-proxy-writer [dlog] > // Output of `TailReader` Opening log stream basic-stream-9 Log stream basic-stream-9 is empty. Wait for records starting from DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} Received record DLSN{logSegmentSequenceNo=1, entryId=0, slotId=0} """ test-proxy-writer """