<commit_before>=  Queue Replication
:toc:
:toc-title: Contents
:toclevels: 2


== Chronicle Queue Enterprise Replication

At the heart of Chronicle Services Replication lies Chronicle Queue Enterprise replication,
Chronicle Queue Enterprise replication performs simple unidirectional multi-way replication by copying a chronicle queue
from one host to one or more others via TCP/IP.

.Chronicle Queue Replication
image::images/queue-replication.png[]

NOTE: for Chronicle-Engine based replication, see link:engine-replication.adoc[here]

=== The mechanics of Chronicle Queue Replication

The queue that we write to we refer to as the source queue and the copy(-ies) we refer to as the sink queue.

IMPORTANT: Chronicle requires exclusive write access to the Sink. Sink should be treated as read-only, and never manually written to.

If we allowed messages to be written to the sink, its not hard to imagine a case where different messages are written to a source
and sink at the same time, the messages on the source queue and sink queue could now be in a very different order,
as such we don’t allow messages to be written to the sync queue. Or putting it another way Queues must maintain message
ordering and the source queue and sink queue are identical. One of the reasons we enforce this is that in a microservices
architecture, in order that you can get true replay-ability you should be able to guarantee the consistency and ordering of
the messages in the queues, hence any service that's running on the source machine must receive the same events,
and in the same order, as the service running on the sink machine.

So in summary, any message written to source is copied to sink. The sink therefore becomes a mirror image of the source.

At startup, replication locks Source queue and waits for Sink(s) to report the number of records it has, and if Sink has
more records, they are replayed back to Source before it is unlocked and usable again. This is done to provide
automatic data re-sync after failover to Sink.

The set of hosts onto which the queue is replicated is defined as a cluster. The config for that looks as follows:

```
!QueueReplicationCfg {
  clusters: !QueueClustersCfg {
    cluster1: {
      context: !QueueClusterContext {
        heartbeatIntervalMs: 300000,
        heartbeatTimeoutMs: 500000,
        baseSourcePath: "replica/source",
        baseSinkPath: "replica/sink",
      }

      host1: {
        hostId: 1,
        connectUri: host.port1,
      }
      host2: {
        hostId: 2,
        connectUri: host.port2,
      }
      ...
    }
  },

  queues: !ReplicatedQueuesCfg {
    queue1: {
      cluster: "cluster1",
      acknowledge: false,
      wireType: !WireType BINARY,
      masterId: 1,
    }
  }
}
```

In the above configuration, the queue "queue1" is set to be replicated from host1 (as indicated by `masterId`) to all other hosts
defined for cluster "cluster1". Queues will use storage paths defined by `baseSourcePath`/`baseSinkPath` for source and
sink, respectively, followed by queue name. So for this example, source queue will be at `replica/source/queue1` while
sink will be written to `replica/sink/queue1`.


=== Using Enterprise Queue Replication

In order to start replicating data, user needs to create an instance of ReplicatedQueue per host. This is done as follows:

```
ReplicatedQueue repl = new ReplicatedQueue(config());
repl.startReplication(hostId, new ServicesReplicationConfig());
...
// shutdown
repl.shutdown();
```
<commit_msg>added source code to docs<commit_after>=  Queue Replication
:toc:
:toc-title: Contents
:toclevels: 2


== Chronicle Queue Enterprise Replication

At the heart of Chronicle Services Replication lies Chronicle Queue Enterprise replication,
Chronicle Queue Enterprise replication performs simple unidirectional multi-way replication by copying a chronicle queue
from one host to one or more others via TCP/IP.

.Chronicle Queue Replication
image::images/queue-replication.png[]

NOTE: for Chronicle-Engine based replication, see link:engine-replication.adoc[here]

=== The mechanics of Chronicle Queue Replication

The queue that we write to we refer to as the source queue and the copy(-ies) we refer to as the sink queue.

IMPORTANT: Chronicle requires exclusive write access to the Sink. Sink should be treated as read-only, and never manually written to.

If we allowed messages to be written to the sink, its not hard to imagine a case where different messages are written to a source
and sink at the same time, the messages on the source queue and sink queue could now be in a very different order,
as such we don’t allow messages to be written to the sync queue. Or putting it another way Queues must maintain message
ordering and the source queue and sink queue are identical. One of the reasons we enforce this is that in a microservices
architecture, in order that you can get true replay-ability you should be able to guarantee the consistency and ordering of
the messages in the queues, hence any service that's running on the source machine must receive the same events,
and in the same order, as the service running on the sink machine.

So in summary, any message written to source is copied to sink. The sink therefore becomes a mirror image of the source.

At startup, replication locks Source queue and waits for Sink(s) to report the number of records it has, and if Sink has
more records, they are replayed back to Source before it is unlocked and usable again. This is done to provide
automatic data re-sync after failover to Sink.

The set of hosts onto which the queue is replicated is defined as a cluster. The config for that looks as follows:

```
!QueueReplicationCfg {
  clusters: !QueueClustersCfg {
    cluster1: {
      context: !QueueClusterContext {
        heartbeatIntervalMs: 300000,
        heartbeatTimeoutMs: 500000,
        baseSourcePath: "replica/source",
        baseSinkPath: "replica/sink",
      }

      host1: {
        hostId: 1,
        connectUri: host.port1,
      }
      host2: {
        hostId: 2,
        connectUri: host.port2,
      }
      ...
    }
  },

  queues: !ReplicatedQueuesCfg {
    queue1: {
      cluster: "cluster1",
      acknowledge: false,
      wireType: !WireType BINARY,
      masterId: 1,
    }
  }
}
```

In the above configuration, the queue "queue1" is set to be replicated from host1 (as indicated by `masterId`) to all other hosts
defined for cluster "cluster1". Queues will use storage paths defined by `baseSourcePath`/`baseSinkPath` for source and
sink, respectively, followed by queue name. So for this example, source queue will be at `replica/source/queue1` while
sink will be written to `replica/sink/queue1`.


=== Using Enterprise Queue Replication

In order to start replicating data, user needs to create an instance of ReplicatedQueue per host. This is done as follows:

```
ReplicatedQueue repl = new ReplicatedQueue(config());
repl.startReplication(hostId, new ServicesReplicationConfig());
...
// shutdown
repl.shutdown();
```


==== Example from a test case

[source,ruby]
```
@Test
public void shouldReplicate() throws Exception {

  YamlLogging.setAll(false);
        IOTools.deleteDirWithFiles("replica", 10);
        TCPRegistry.createServerSocketChannelFor(
                "host.port1",
                "host.port2",
                "host.port3");

    startupHost((byte) 1);
    startupHost((byte) 2);

    String queueName = config().queues().getQueues().iterator().next();
    ReplicatedQueueCfg qCfg = config().queues().getQueue(queueName);
    QueueClusterContext clusterContext = config().clusters().getClusters().iterator().next().clusterContext();
    String sourceBasePath = clusterContext.baseSourcePath();
    String sinkBasePath = clusterContext.baseSinkPath();
    WireType wireType = qCfg.wireType();
    SingleChronicleQueue source = SingleChronicleQueueBuilder.builder(Paths.get(sourceBasePath, queueName), wireType).build();
    SingleChronicleQueue sink = SingleChronicleQueueBuilder.builder(Paths.get(sinkBasePath, queueName), wireType).build();

    final BlockingQueue<String> sinkMessages = new LinkedBlockingQueue<>();
    ChronicleReader chronicleReader = new ChronicleReader().withBasePath(sink.file().toPath()).withMessageSink(msg -> {
        if (!msg.startsWith("0x"))
            sinkMessages.add(msg);
    }).tail();
    Thread readerThread = new Thread(chronicleReader::execute);
    readerThread.start();

    assertNull(sinkMessages.poll(200L, TimeUnit.MILLISECONDS));

    ExcerptAppender appender = source.acquireAppender();

    try (DocumentContext dc = appender.writingDocument()) {
        dc.wire().write("test").text("Hello replica");
    }

    String poll = sinkMessages.poll(5000L, TimeUnit.MILLISECONDS);
    assertEquals("test: Hello replica\n", poll);

    try (DocumentContext dc = appender.writingDocument()) {
        dc.wire().write("test2").text("Hello replica");
    }

    poll = sinkMessages.poll(5000L, TimeUnit.MILLISECONDS);
    assertEquals("test2: Hello replica\n", poll);

    poll = sinkMessages.poll(500L, TimeUnit.MILLISECONDS);
    assertNull(poll);
    readerThread.interrupt();

    readerThread.join();
}
```
{'fraction_non_alphanumeric': 0.07704026115342764, 'fraction_numerical': 0.008705114254624592, 'mean_word_length': 3.8888297872340427, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2809732', 'n_tokens_mistral': 2571, 'n_tokens_neox': 2443, 'n_words': 1061}