---
title: API - Rewind reading records by time
layout: default
---

.. contents:: Basic Tutorial - Rewind reading records by time

Rewind reading records by time
==============================

This tutorial shows how to rewind reading data from a stream by time.

.. sectnum::

Open a distributedlog manager
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Create distributedlog URI
-------------------------

::

        String dlUriStr = ...;
        URI uri = URI.create(dlUriStr);


Create distributedlog configuration
-----------------------------------

::

        DistributedLogConfiguration conf = new DistributedLogConfiguration();


Build the distributedlog namespace
----------------------------------

::

        DistributedLogNamespace namespace = DistributedLogNamespaceBuilder.newBuilder()
                .conf(conf)
                .uri(uri)
                .build(); 


Open the distributedlog manager
-------------------------------

::

        DistributedLogManager dlm = namespace.openLog("basic-stream-10");


Rewind the stream
~~~~~~~~~~~~~~~~~

Position the reader using timestamp
-----------------------------------

Since the records written by write proxy will be assigned `System.currentTimeMillis()`
as the `TransactionID`. It is straightforward to use `TransactionID` to rewind reading
the records.

::

        int rewindSeconds = 60; // 60 seconds
        long fromTxID = System.currentTimeMillis() -
                TimeUnit.MILLISECONDS.convert(rewindSeconds, TimeUnit.SECONDS);
        AsyncLogReader reader = FutureUtils.result(dlm.openAsyncLogReader(fromTxID));


Read Records
~~~~~~~~~~~~

Read the next available record from the stream. The future is satisified when the record is available.

::

        Future<LogRecordWithDLSN> readFuture = reader.readNext();


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


Close the reader
~~~~~~~~~~~~~~~~

Close the reader after usage.

::

        FutureUtils.result(reader.asyncClose());


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


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 the stream
-----------------

Create the stream under the distributedlog uri.

::

        // Create Stream `basic-stream-10`
        // 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 10


Generate records
----------------

Run the `RecordGenerator` to generate records.

::

        // Write Records into Stream `basic-stream-10` in 1 requests/second
        // runner run org.apache.distributedlog.basic.RecordGenerator ${distributedlog-uri} ${stream} ${rate}
        ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.RecordGenerator 'inet!127.0.0.1:8000' basic-stream-10 1


Rewind the stream
-----------------

Rewind the stream using `StreamRewinder` to read records from 30 seconds ago

::

        // Rewind `basic-stream-10`
        // runner run org.apache.distributedlog.basic.StreamRewinder ${distributedlog-uri} ${stream} ${seconds-to-rewind}
        ./distributedlog-tutorials/distributedlog-basic/bin/runner run org.apache.distributedlog.basic.StreamRewinder distributedlog://127.0.0.1:7000/messaging/distributedlog basic-stream-10  30


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

Example output from `StreamRewinder`.

::

        // Output of `StreamRewinder`
        Opening log stream basic-stream-10
        Record records starting from 1462736697481 which is 30 seconds ago
        Received record DLSN{logSegmentSequenceNo=1, entryId=264, slotId=0}
        """
        record-1462736697685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=266, slotId=0}
        """
        record-1462736698684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=268, slotId=0}
        """
        record-1462736699684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=270, slotId=0}
        """
        record-1462736700686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=272, slotId=0}
        """
        record-1462736701685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=274, slotId=0}
        """
        record-1462736702684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=276, slotId=0}
        """
        record-1462736703683
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=278, slotId=0}
        """
        record-1462736704685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=280, slotId=0}
        """
        record-1462736705686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=282, slotId=0}
        """
        record-1462736706682
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=284, slotId=0}
        """
        record-1462736707685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=286, slotId=0}
        """
        record-1462736708686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=288, slotId=0}
        """
        record-1462736709684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=290, slotId=0}
        """
        record-1462736710684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=292, slotId=0}
        """
        record-1462736711686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=294, slotId=0}
        """
        record-1462736712686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=296, slotId=0}
        """
        record-1462736713684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=298, slotId=0}
        """
        record-1462736714682
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=300, slotId=0}
        """
        record-1462736715685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=302, slotId=0}
        """
        record-1462736716684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=304, slotId=0}
        """
        record-1462736717684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=306, slotId=0}
        """
        record-1462736718684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=308, slotId=0}
        """
        record-1462736719685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=310, slotId=0}
        """
        record-1462736720683
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=312, slotId=0}
        """
        record-1462736721686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=314, slotId=0}
        """
        record-1462736722685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=316, slotId=0}
        """
        record-1462736723683
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=318, slotId=0}
        """
        record-1462736724683
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=320, slotId=0}
        """
        record-1462736725685
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=322, slotId=0}
        """
        record-1462736726686
        """
        Reader caught with latest data
        Received record DLSN{logSegmentSequenceNo=1, entryId=324, slotId=0}
        """
        record-1462736727686
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=326, slotId=0}
        """
        record-1462736728684
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=328, slotId=0}
        """
        record-1462736729682
        """
        Received record DLSN{logSegmentSequenceNo=1, entryId=330, slotId=0}
        """
        record-1462736730685
        """