Repository: liftbridge-io/liftbridge
Pull request #107 (2019-10-16 18:48:41)

Fix idle follower ack delay
Fix issue #102 by having the leader notify followers when there is new
data available after the follower has caught up with the leader's log
end offset.

This also introduces a new config `replica.max.idle.wait` which
controls the max amount of time an idle follower will wait before making
a replication request. The primary purpose of this is to ensure timely
check-ins with the leader for health checking.


Patch:
diff --git a/documentation/configuration.md b/documentation/configuration.md
index fb94ca2..df364ae 100644
--- a/documentation/configuration.md
+++ b/documentation/configuration.md
@@ -158,7 +158,8 @@ the configuration file.
 | raft.bootstrap.seed | raft-bootstrap-seed | Bootstrap the Raft cluster by electing self as leader if there is no existing state. | bool | false | |
 | raft.bootstrap.peers | raft-bootstrap-peers | Bootstrap the Raft cluster with the provided list of peer IDs if there is no existing state. | list | | |
 | raft.logging | | Enables logging in the Raft subsystem. | bool | false | |
-| replica.max.lag.time | | If a follower hasn't sent any replication requests or hasn't caught up to the leader's log end offset for at least this time, the leader will remove the follower from ISR. | duration | 10s | |
-| replica.max.leader.timeout | | If a leader hasn't sent any replication responses for at least this time, the follower will report the leader to the controller. If a majority of the replicas report the leader, a new leader is selected by the controller. | duration | 10s | |
+| replica.max.lag.time | | If a follower hasn't sent any replication requests or hasn't caught up to the leader's log end offset for at least this time, the leader will remove the follower from ISR. | duration | 15s | |
+| replica.max.leader.timeout | | If a leader hasn't sent any replication responses for at least this time, the follower will report the leader to the controller. If a majority of the replicas report the leader, a new leader is selected by the controller. | duration | 15s | |
+| replica.max.idle.wait | | The maximum amount of time a follower will wait before making a replication request once the follower is caught up with the leader. This value should always be less than `replica.max.lag.time` to avoid frequent shrinking of ISR for low-throughput streams. | duration | 10s | |
 | replica.fetch.timeout | | Timeout duration for follower replication requests. | duration | 3s | |
 | min.insync.replicas | | Specifies the minimum number of replicas that must acknowledge a stream write before it can be committed. If the ISR drops below this size, messages cannot be committed. | int | 1 | [1,...] |
diff --git a/documentation/replication_protocol.md b/documentation/replication_protocol.md
index 8e7541f..04ff7c2 100644
--- a/documentation/replication_protocol.md
+++ b/documentation/replication_protocol.md
@@ -26,22 +26,38 @@ follows:
    recovery purposes described below.
 1. The nodes participating in the partition initialize it, and the leader
    subscribes to the NATS subject.
-1. The leader initializes the high watermark (HW) to -1. This is the offset of
+1. The leader initializes the high watermark (`HW`) to -1. This is the offset of
    the last committed message in the partition (-1 meaning, initially, no
    messages are committed).
 1. The leader begins sequencing messages from NATS and writes them to the log
    uncommitted.
 1. Followers consume from the leader's log to replicate messages to their own
-   log. We piggyback the leader's HW and `LeaderEpoch` on these responses.
+   log. We piggyback the leader's `HW` and `LeaderEpoch` on these responses.
    Replicas update their `LeaderEpoch` cache as needed and periodically
-   checkpoint the HW to stable storage.
-1. Replicas acknowledge they've replicated the message via replication requests
-   to the leader.
-1. Once the leader has heard from the ISR, the message is committed and the HW
-   is updated.
+   checkpoint the `HW` to stable storage. If the `LeaderEpoch` on the response
+   does not match the expected `LeaderEpoch`, the follower drops the response.
+   The same is true if the offset of the first message in the response is less
+   than the expected next offset. The follower uses the leader's `HW` to update
+   their own `HW`.
+1. Followers acknowledge they've replicated the message via replication
+   requests to the leader.
+1. Once the leader has heard from the ISR, the message is committed and the
+   `HW` is updated.
 
 Note that clients only see committed messages in the log.
 
+When a follower has caught up with the leader's log (i.e. it reaches the
+leader's log end offset, or LEO), the leader will register a _data waiter_ for
+the follower on the log. This waiter is used to signal when new messages have
+been written to the log. Upon catching up, followers will receive an empty
+replication response from the leader and sleep for a duration
+`replica.max.idle.wait` to avoid overloading the leader with replication
+requests. After sleeping, they will make subsequent replication requests. Even
+if data is not available, this will ensure timely health checks occur to avoid
+spurious ISR shrinking. If new messages are written while a follower is idle,
+the data waiter is signalled which causes the leader to send a notification to
+the follower to preempt the sleep and begin replicating again.
+
 ## Failure Modes
 
 There are a variety of failures that can occur in the replication process. A
@@ -75,14 +91,55 @@ replicates this fact via Raft. The partition leader continues to commit messages
 with fewer replicas in the ISR, entering an under-replicated state.
 
 When a failed follower is restarted, it requests the last offset for the
-current `LeaderEpoch` from the partition leader. It then truncates its log up to
-this offset, which removes any potentially uncommitted messages from the log.
-The replica then begins fetching messages from the leader starting at this
-offset. Once the replica has caught up, it's added back into the ISR and the
-system resumes its fully replicated state. Adding a replica back into the
-ISR results in incrementing the partition's `Epoch`.
+current `LeaderEpoch` from the partition leader, called a `LeaderEpoch` offset
+request. It then truncates its log up to this offset, which removes any
+potentially uncommitted messages from the log.  The replica then begins
+fetching messages from the leader starting at this offset. Once the replica has
+caught up, it's added back into the ISR and the system resumes its fully
+replicated state. Adding a replica back into the ISR results in incrementing
+the partition's `Epoch`.
 
 We truncate uncommitted messages from the log using the `LeaderEpoch` rather
 than the HW because there are certain edge cases that can result in data loss
-or divergent log lineages with the latter. This has been [described in detail](https://cwiki.apache.org/confluence/display/KAFKA/KIP-101+-+Alter+Replication+Protocol+to+use+Leader+Epoch+rather+than+High+Watermark+for+Truncation)
+or divergent log lineages with the latter. This has been [described in
+detail](https://cwiki.apache.org/confluence/display/KAFKA/KIP-101+-+Alter+Replication+Protocol+to+use+Leader+Epoch+rather+than+High+Watermark+for+Truncation)
 by the maintainers of Kafka.
+
+## Replication RPC Protocol
+
+Replication RPCs are made over internal NATS subjects. Replication requests for
+a partition are sent to `<namespace>.<stream>.<partition>.replicate` which is a
+subject the partition leader subscribes to. The request data is a
+[protobuf](https://github.com/liftbridge-io/liftbridge/blob/8bee0478da97711dc2a8e1fdae8b2d2e3086c756/server/proto/internal.proto#L87-L90)
+containing the ID of the follower and the offset they want to begin fetching
+from. The NATS message also includes a random [reply
+inbox](https://nats-io.github.io/docs/developer/sending/replyto.html) the
+leader uses to send the response to.
+
+The leader response uses a binary format consisting of the following:
+
+```
+   0     1     2     3     4     5     6     7     8     9     10    11    12    13    14    15
++-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+...
+|                  LeaderEpoch                  |                       HW                      |...
++-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+-----+...
+|<-------------------8 bytes------------------->|<-------------------8 bytes------------------->|
+```
+
+The remainder of the response consists of message data. If the follower is
+caught up, there won't be any message data. The `LeaderEpoch` and `HW` are
+always present, so there are 16 bytes guaranteed in the response.
+
+The `LeaderEpoch` offset requests also use internal NATS subjects similar to
+replication RPCs. These requests are sent to
+`<namespace>.<stream>.<partition>.offset`. The request and response both use a
+[protobuf](https://github.com/liftbridge-io/liftbridge/blob/8bee0478da97711dc2a8e1fdae8b2d2e3086c756/server/proto/internal.proto#L92-L98)
+containing the `LeaderEpoch` and offset, respectively. Like replication,
+responses are sent to a random reply subject included on the NATS request
+message.
+
+Notifications of new data are sent on the NATS subject
+`<namespace>.notify.<serverID>`. This is also a protobuf containing the stream
+name and ID of the partition with new data available. Upon receiving this
+notification, the follower preempts the replication loop for the respective
+partition, if idle, to begin replicating.
diff --git a/server/commitlog.go b/server/commitlog.go
index 3a46807..daeb9e7 100644
--- a/server/commitlog.go
+++ b/server/commitlog.go
@@ -61,6 +61,13 @@ type CommitLog interface {
 	// applicable.
 	Clean() error
 
+	// NotifyLEO registers and returns a channel which is closed when messages
+	// past the given log end offset are added to the log. If the given offset
+	// is no longer the log end offset, the channel is closed immediately.
+	// Waiter is an opaque value that uniquely identifies the entity waiting
+	// for data.
+	NotifyLEO(waiter interface{}, leo int64) <-chan struct{}
+
 	// Close closes each log segment file and stops the background goroutine
 	// checkpointing the high watermark to disk.
 	Close() error
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index 10185d1..d27dd02 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -246,11 +246,14 @@ func (l *CommitLog) append(segment *Segment, ms []byte, entries []*Entry) ([]int
 	if err := segment.WriteMessageSet(ms, entries); err != nil {
 		return nil, err
 	}
-	lastLeaderEpoch := l.leaderEpochCache.LastLeaderEpoch()
-	offsets := make([]int64, len(entries))
+	var (
+		lastLeaderEpoch = l.leaderEpochCache.LastLeaderEpoch()
+		offsets         = make([]int64, len(entries))
+	)
 	for i, entry := range entries {
 		// Check if message is in a new leader epoch.
 		if entry.LeaderEpoch > lastLeaderEpoch {
+			// If it is, we need to assign the epoch offset.
 			if err := l.leaderEpochCache.Assign(entry.LeaderEpoch, entry.Offset); err != nil {
 				return nil, err
 			}
@@ -502,6 +505,14 @@ func (l *CommitLog) Segments() []*Segment {
 	return l.segments
 }
 
+// NotifyLEO registers and returns a channel which is closed when messages past
+// the given log end offset are added to the log. If the given offset is no
+// longer the log end offset, the channel is closed immediately. Waiter is an
+// opaque value that uniquely identifies the entity waiting for data.
+func (l *CommitLog) NotifyLEO(waiter interface{}, leo int64) <-chan struct{} {
+	return l.activeSegment().WaitForLEO(waiter, leo)
+}
+
 // checkAndPerformSplit determines if a new log segment should be rolled out
 // either because the active segment is full or LogRollTime has passed since
 // the first message was written to it. It then performs the split if eligible,
diff --git a/server/commitlog/reader.go b/server/commitlog/reader.go
index c10a447..9e759a2 100644
--- a/server/commitlog/reader.go
+++ b/server/commitlog/reader.go
@@ -157,7 +157,7 @@ LOOP:
 }
 
 func (r *uncommittedReader) waitForData(ctx context.Context, seg *Segment) bool {
-	wait := seg.waitForData(r, r.pos)
+	wait := seg.WaitForData(r, r.pos)
 	select {
 	case <-r.cl.closed:
 		seg.removeWaiter(r)
diff --git a/server/commitlog/segment.go b/server/commitlog/segment.go
index d6656ce..62cc112 100644
--- a/server/commitlog/segment.go
+++ b/server/commitlog/segment.go
@@ -57,7 +57,7 @@ type Segment struct {
 	maxBytes       int64
 	path           string
 	suffix         string
-	waiters        map[contextReader]chan struct{}
+	waiters        map[interface{}]chan struct{}
 	sealed         bool
 	closed         bool
 	replaced       bool
@@ -73,7 +73,7 @@ func NewSegment(path string, baseOffset, maxBytes int64, isNew bool, suffix stri
 		lastOffset:  -1,
 		path:        path,
 		suffix:      suffix,
-		waiters:     make(map[contextReader]chan struct{}),
+		waiters:     make(map[interface{}]chan struct{}),
 	}
 	// If this is a new segment, ensure the file doesn't already exist.
 	if isNew && exists(s.logPath()) {
@@ -252,22 +252,42 @@ func (s *Segment) notifyWaiters() {
 	}
 }
 
-func (s *Segment) waitForData(r contextReader, pos int64) <-chan struct{} {
-	wait := make(chan struct{})
+func (s *Segment) WaitForLEO(waiter interface{}, leo int64) <-chan struct{} {
 	s.Lock()
+	defer s.Unlock()
+	if s.lastOffset != leo {
+		ch := make(chan struct{})
+		close(ch)
+		return ch
+	}
+	return s.waitForData(waiter, s.position)
+}
+func (s *Segment) WaitForData(waiter interface{}, pos int64) <-chan struct{} {
+	s.Lock()
+	ch := s.waitForData(waiter, pos)
+	s.Unlock()
+	return ch
+}
+
+func (s *Segment) waitForData(waiter interface{}, pos int64) <-chan struct{} {
+	// Check if we're already registered.
+	wait, ok := s.waiters[waiter]
+	if ok {
+		return wait
+	}
+	wait = make(chan struct{})
 	// Check if data has been written and/or the segment was filled.
 	if s.position > pos || s.position >= s.maxBytes {
 		close(wait)
 	} else {
-		s.waiters[r] = wait
+		s.waiters[waiter] = wait
 	}
-	s.Unlock()
 	return wait
 }
 
-func (s *Segment) removeWaiter(r contextReader) {
+func (s *Segment) removeWaiter(waiter interface{}) {
 	s.Lock()
-	delete(s.waiters, r)
+	delete(s.waiters, waiter)
 	s.Unlock()
 }
 
diff --git a/server/config.go b/server/config.go
index 7794896..b279aac 100644
--- a/server/config.go
+++ b/server/config.go
@@ -28,8 +28,9 @@ const (
 const (
 	defaultListenAddress           = "0.0.0.0"
 	defaultConnectionAddress       = "localhost"
-	defaultReplicaMaxLagTime       = 10 * time.Second
-	defaultReplicaMaxLeaderTimeout = 10 * time.Second
+	defaultReplicaMaxLagTime       = 15 * time.Second
+	defaultReplicaMaxLeaderTimeout = 15 * time.Second
+	defaultReplicaMaxIdleWait      = 10 * time.Second
 	defaultRaftSnapshots           = 2
 	defaultRaftCacheSize           = 512
 	defaultMetadataCacheMaxAge     = 2 * time.Minute
@@ -93,6 +94,7 @@ type ClusteringConfig struct {
 	ReplicaMaxLagTime       time.Duration
 	ReplicaMaxLeaderTimeout time.Duration
 	ReplicaFetchTimeout     time.Duration
+	ReplicaMaxIdleWait      time.Duration
 	MinISR                  int
 }
 
@@ -130,6 +132,7 @@ func NewDefaultConfig() *Config {
 	config.Clustering.Namespace = DefaultNamespace
 	config.Clustering.ReplicaMaxLagTime = defaultReplicaMaxLagTime
 	config.Clustering.ReplicaMaxLeaderTimeout = defaultReplicaMaxLeaderTimeout
+	config.Clustering.ReplicaMaxIdleWait = defaultReplicaMaxIdleWait
 	config.Clustering.ReplicaFetchTimeout = defaultReplicaFetchTimeout
 	config.Clustering.RaftSnapshots = defaultRaftSnapshots
 	config.Clustering.RaftCacheSize = defaultRaftCacheSize
@@ -379,6 +382,12 @@ func parseClusteringConfig(config *Config, m map[string]interface{}) error {
 				return err
 			}
 			config.Clustering.ReplicaMaxLeaderTimeout = dur
+		case "replica.max.idle.wait":
+			dur, err := time.ParseDuration(v.(string))
+			if err != nil {
+				return err
+			}
+			config.Clustering.ReplicaMaxIdleWait = dur
 		case "replica.fetch.timeout":
 			dur, err := time.ParseDuration(v.(string))
 			if err != nil {
diff --git a/server/metadata.go b/server/metadata.go
index cdb9c65..8355565 100644
--- a/server/metadata.go
+++ b/server/metadata.go
@@ -202,7 +202,7 @@ func (m *metadataAPI) fetchBrokerInfo(ctx context.Context, numPeers int) ([]*cli
 	if err != nil {
 		panic(err)
 	}
-	m.ncRaft.PublishRequest(m.serverInfoInbox(), inbox, queryReq)
+	m.ncRaft.PublishRequest(m.getServerInfoInbox(), inbox, queryReq)
 
 	// Gather responses.
 	for i := 0; i < numPeers; i++ {
@@ -735,7 +735,7 @@ func (m *metadataAPI) waitForPartitionLeader(ctx context.Context, stream, leader
 	if err != nil {
 		panic(err)
 	}
-	inbox := fmt.Sprintf(partitionStatusInboxTemplate, m.baseMetadataRaftSubject(), leader)
+	inbox := m.getPartitionStatusInbox(leader)
 	for i := 0; i < 5; i++ {
 		resp, err := m.ncRaft.RequestWithContext(ctx, inbox, req)
 		if err != nil {
diff --git a/server/partition.go b/server/partition.go
index 76f9306..853d053 100644
--- a/server/partition.go
+++ b/server/partition.go
@@ -3,6 +3,7 @@ package server
 import (
 	"bytes"
 	"fmt"
+	"math/rand"
 	"path/filepath"
 	"strconv"
 	"sync"
@@ -87,6 +88,7 @@ type partition struct {
 	recovered       bool
 	stopFollower    chan struct{}
 	stopLeader      chan struct{}
+	notify          chan struct{}
 	belowMinISR     bool
 	pause           bool // Pause replication on the leader (for unit testing)
 	shutdown        sync.WaitGroup
@@ -145,6 +147,7 @@ func (s *Server) newPartition(protoPartition *proto.Partition, recovered bool) (
 		replicas:    replicas,
 		isr:         isr,
 		commitCheck: make(chan struct{}, len(protoPartition.Replicas)),
+		notify:      make(chan struct{}, 1),
 		recovered:   recovered,
 	}
 
@@ -178,6 +181,25 @@ func (p *partition) Close() error {
 	return nil
 }
 
+// Notify is used to short circuit the sleep backoff a partition uses when it
+// has replicated to the end of the leader's log (i.e. the log end offset).
+// When a follower reaches the end of the log, it starts to sleep in between
+// replication requests to avoid overloading the leader. However, this causes
+// added commit latency when new messages are published to the log since the
+// follower is idle. As a result, the leader will note when a follower is
+// caught up and send a notification in order to wake an idle follower back up
+// when new data is written to the log.
+func (p *partition) Notify() {
+	if p.IsLeader() {
+		// If we are now the leader, do nothing.
+		return
+	}
+	select {
+	case p.notify <- struct{}{}:
+	default:
+	}
+}
+
 // SetLeader sets the leader for the partition to the given replica and leader
 // epoch. If the partition's current leader epoch is greater than the given
 // epoch, this returns an error. This will also start the partition as a leader
@@ -424,6 +446,7 @@ func (p *partition) handleLeaderOffsetRequest(msg *nats.Msg) {
 // receives a replication request from a follower. It will send messages to the
 // NATS subject specified on the request.
 func (p *partition) handleReplicationRequest(msg *nats.Msg) {
+	received := time.Now()
 	req := &proto.ReplicationRequest{}
 	if err := req.Unmarshal(msg.Data); err != nil {
 		p.srv.logger.Errorf("Invalid replication request for partition %s: %v", p, err)
@@ -445,15 +468,16 @@ func (p *partition) handleReplicationRequest(msg *nats.Msg) {
 		panic(fmt.Sprintf("No replicator for partition %s and replica %s", p, req.ReplicaID))
 	}
 	p.mu.Unlock()
-	replicator.request(replicationRequest{req, msg})
+	replicator.request(replicationRequest{req, msg, received})
 }
 
 // handleReplicationResponse is a NATS handler that's invoked when a follower
 // receives a replication response from the leader. This response will contain
 // the leader epoch, leader HW, and (optionally) messages to replicate.
 func (p *partition) handleReplicationResponse(msg *nats.Msg) int {
-	// We should have at least 16 bytes, 8 for leader epoch and 8 for HW.
-	if len(msg.Data) < 16 {
+	// We should have at least 16 bytes, 8 for leader epoch and 8 for HW, i.e.
+	// replicationOverhead.
+	if len(msg.Data) < replicationOverhead {
 		p.srv.logger.Warnf("Invalid replication response for partition %s", p)
 		return 0
 	}
@@ -476,7 +500,7 @@ func (p *partition) handleReplicationResponse(msg *nats.Msg) int {
 	hw := int64(proto.Encoding.Uint64(msg.Data[8:]))
 	p.log.SetHighWatermark(hw)
 
-	data := msg.Data[16:]
+	data := msg.Data[replicationOverhead:]
 	if len(data) == 0 {
 		return 0
 	}
@@ -628,16 +652,10 @@ func (p *partition) startReplicating(epoch uint64, stop chan struct{}) {
 			// Don't replicate to ourselves.
 			continue
 		}
-		r := &replicator{
-			replica:    replica,
-			partition:  p,
-			requests:   make(chan replicationRequest, 1),
-			maxLagTime: p.srv.config.Clustering.ReplicaMaxLagTime,
-			leader:     p.srv.config.Clustering.ServerID,
-		}
+		r := newReplicator(epoch, replica, p)
 		p.replicators[replica] = r
 		p.srv.startGoroutine(func() {
-			r.start(epoch, stop)
+			r.start(stop)
 			p.shutdown.Done()
 		})
 	}
@@ -728,11 +746,8 @@ func (p *partition) sendAck(ack *client.Ack) {
 // replicationRequestLoop is a long-running loop which sends replication
 // requests to the partition leader, handles replicating messages, and checks
 // the health of the leader.
-func (p *partition) replicationRequestLoop(leader string, epoch uint64, stop chan struct{}) {
-	var (
-		emptyResponseCount int
-		leaderLastSeen     = time.Now()
-	)
+func (p *partition) replicationRequestLoop(leader string, epoch uint64, stop <-chan struct{}) {
+	leaderLastSeen := time.Now()
 	for {
 		select {
 		case <-stop:
@@ -748,22 +763,25 @@ func (p *partition) replicationRequestLoop(leader string, epoch uint64, stop cha
 			leaderLastSeen = time.Now()
 		}
 
-		select {
-		case <-stop:
-			return
-		default:
-		}
-
 		// Check if leader has exceeded max leader timeout.
 		p.checkLeaderHealth(leader, epoch, leaderLastSeen)
 
-		// TODO: Make this smarter. Probably have the request wait for data on
-		// the server side. Need to handle timeouts better too.
-		if replicated == 0 {
-			time.Sleep(computeReplicaFetchSleep(emptyResponseCount))
-			emptyResponseCount++
-		} else {
-			emptyResponseCount = 0
+		// If there is more data or we errored, continue replicating.
+		if replicated > 0 || err != nil {
+			continue
+		}
+
+		// If we are caught up with the leader, wait for data.
+		wait := p.computeReplicaFetchSleep()
+		select {
+		case <-stop:
+			return
+		case <-time.After(wait):
+			// Check in with leader to maintain health status.
+			continue
+		case <-p.notify:
+			// Leader has signalled more data is available.
+			continue
 		}
 	}
 }
@@ -793,16 +811,16 @@ func (p *partition) checkLeaderHealth(leader string, epoch uint64, leaderLastSee
 
 // computeReplicaFetchSleep calculates the time to backoff before sending
 // another replication request.
-func computeReplicaFetchSleep(emptyResponseCount int) time.Duration {
-	sleep := time.Duration(300+emptyResponseCount*5) * time.Millisecond
-	if sleep > time.Second {
-		sleep = time.Second
-	}
-	return sleep
+func (p *partition) computeReplicaFetchSleep() time.Duration {
+	sleep := p.srv.config.Clustering.ReplicaMaxIdleWait
+	// Subtract some random jitter from the max wait time.
+	return sleep - time.Duration(rand.Intn(2000))*time.Millisecond
 }
 
 // sendReplicationRequest sends a replication request to the partition leader
-// and processes the response.
+// and processes the response. It returns an int indicating the number of
+// messages that were replicated. Zero (without an error) indicates the
+// follower is caught up with the leader.
 func (p *partition) sendReplicationRequest() (int, error) {
 	data, err := (&proto.ReplicationRequest{
 		ReplicaID: p.srv.config.Clustering.ServerID,
@@ -819,8 +837,7 @@ func (p *partition) sendReplicationRequest() (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	replicated := p.handleReplicationResponse(resp)
-	return replicated, nil
+	return p.handleReplicationResponse(resp), nil
 }
 
 // truncateUncommitted truncates the log up to the start offset of the first
@@ -1050,6 +1067,19 @@ func (p *partition) updateISRLatestOffset(replica string, offset int64) {
 	}
 }
 
+// sendPartitionNotification sends a message to the given partition replica to
+// indicate new data is available in the log.
+func (p *partition) sendPartitionNotification(replica string) {
+	req, err := (&proto.PartitionNotification{
+		Stream:    p.Stream,
+		Partition: p.Id,
+	}).Marshal()
+	if err != nil {
+		panic(err)
+	}
+	p.srv.ncRepl.Publish(p.srv.getPartitionNotificationInbox(replica), req)
+}
+
 // pauseReplication stops replication on the leader. This is for unit testing
 // purposes only.
 func (p *partition) pauseReplication() {
diff --git a/server/proto/internal.pb.go b/server/proto/internal.pb.go
index afe496e..7b06fa8 100644
--- a/server/proto/internal.pb.go
+++ b/server/proto/internal.pb.go
@@ -29,6 +29,7 @@
 		ServerInfoResponse
 		PartitionStatusRequest
 		PartitionStatusResponse
+		PartitionNotification
 */
 package proto
 
@@ -733,6 +734,30 @@ func (m *PartitionStatusResponse) GetIsLeader() bool {
 	return false
 }
 
+type PartitionNotification struct {
+	Stream    string `protobuf:"bytes,1,opt,name=stream,proto3" json:"stream,omitempty"`
+	Partition int32  `protobuf:"varint,2,opt,name=partition,proto3" json:"partition,omitempty"`
+}
+
+func (m *PartitionNotification) Reset()                    { *m = PartitionNotification{} }
+func (m *PartitionNotification) String() string            { return proto1.CompactTextString(m) }
+func (*PartitionNotification) ProtoMessage()               {}
+func (*PartitionNotification) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{21} }
+
+func (m *PartitionNotification) GetStream() string {
+	if m != nil {
+		return m.Stream
+	}
+	return ""
+}
+
+func (m *PartitionNotification) GetPartition() int32 {
+	if m != nil {
+		return m.Partition
+	}
+	return 0
+}
+
 func init() {
 	proto1.RegisterType((*ServerState)(nil), "proto.ServerState")
 	proto1.RegisterType((*RaftLog)(nil), "proto.RaftLog")
@@ -755,6 +780,7 @@ func init() {
 	proto1.RegisterType((*ServerInfoResponse)(nil), "proto.ServerInfoResponse")
 	proto1.RegisterType((*PartitionStatusRequest)(nil), "proto.PartitionStatusRequest")
 	proto1.RegisterType((*PartitionStatusResponse)(nil), "proto.PartitionStatusResponse")
+	proto1.RegisterType((*PartitionNotification)(nil), "proto.PartitionNotification")
 	proto1.RegisterEnum("proto.Op", Op_name, Op_value)
 }
 func (m *ServerState) Marshal() (dAtA []byte, err error) {
@@ -1547,6 +1573,35 @@ func (m *PartitionStatusResponse) MarshalTo(dAtA []byte) (int, error) {
 	return i, nil
 }
 
+func (m *PartitionNotification) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalTo(dAtA)
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *PartitionNotification) MarshalTo(dAtA []byte) (int, error) {
+	var i int
+	_ = i
+	var l int
+	_ = l
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
+	}
+	if m.Partition != 0 {
+		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
+	}
+	return i, nil
+}
+
 func encodeVarintInternal(dAtA []byte, offset int, v uint64) int {
 	for v >= 1<<7 {
 		dAtA[offset] = uint8(v&0x7f | 0x80)
@@ -1906,6 +1961,19 @@ func (m *PartitionStatusResponse) Size() (n int) {
 	return n
 }
 
+func (m *PartitionNotification) Size() (n int) {
+	var l int
+	_ = l
+	l = len(m.Stream)
+	if l > 0 {
+		n += 1 + l + sovInternal(uint64(l))
+	}
+	if m.Partition != 0 {
+		n += 1 + sovInternal(uint64(m.Partition))
+	}
+	return n
+}
+
 func sovInternal(x uint64) (n int) {
 	for {
 		n++
@@ -4533,6 +4601,104 @@ func (m *PartitionStatusResponse) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *PartitionNotification) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowInternal
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= (uint64(b) & 0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: PartitionNotification: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: PartitionNotification: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Stream", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowInternal
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= (uint64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthInternal
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Stream = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Partition", wireType)
+			}
+			m.Partition = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowInternal
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.Partition |= (int32(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		default:
+			iNdEx = preIndex
+			skippy, err := skipInternal(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthInternal
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func skipInternal(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
@@ -4641,62 +4807,63 @@ var (
 func init() { proto1.RegisterFile("server/proto/internal.proto", fileDescriptorInternal) }
 
 var fileDescriptorInternal = []byte{
-	// 902 bytes of a gzipped FileDescriptorProto
+	// 915 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xcc, 0x56, 0x4f, 0x73, 0x1b, 0x35,
-	0x14, 0xef, 0xae, 0xe3, 0x24, 0xfb, 0xdc, 0xba, 0x6b, 0x4d, 0x09, 0x5b, 0xc8, 0x64, 0x3c, 0xe2,
-	0x12, 0x18, 0x48, 0x99, 0xc0, 0x85, 0x19, 0x38, 0x98, 0x64, 0x4b, 0x5d, 0x5c, 0xdb, 0x23, 0xfb,
-	0xc0, 0x89, 0xa0, 0x7a, 0x15, 0x7b, 0x21, 0x59, 0x09, 0x49, 0xe9, 0xf4, 0x5b, 0x70, 0x84, 0xe1,
-	0xd6, 0x13, 0x5f, 0x85, 0x23, 0x1f, 0x81, 0x09, 0x5f, 0x84, 0x91, 0x56, 0xfb, 0xc7, 0xeb, 0x0e,
-	0x33, 0xb8, 0x97, 0x9e, 0xac, 0xdf, 0xd3, 0xd3, 0x7b, 0xbf, 0xf7, 0x7b, 0xda, 0x27, 0xc3, 0xfb,
-	0x8a, 0xc9, 0x17, 0x4c, 0x3e, 0x12, 0x92, 0x6b, 0xfe, 0x28, 0xcd, 0x34, 0x93, 0x19, 0xbd, 0x3a,
-	0xb1, 0x10, 0xb5, 0xed, 0x0f, 0xfe, 0x10, 0x3a, 0x33, 0xeb, 0x35, 0xd3, 0x54, 0x33, 0xf4, 0x1e,
-	0xec, 0xe7, 0x87, 0x86, 0xe7, 0x91, 0xd7, 0xf7, 0x8e, 0x03, 0x52, 0x62, 0xfc, 0xab, 0x0f, 0x7b,
-	0x84, 0x5e, 0xea, 0x11, 0x5f, 0xa2, 0x87, 0xe0, 0x73, 0x61, 0x3d, 0xba, 0xa7, 0x41, 0x1e, 0xf1,
-	0x64, 0x22, 0x88, 0xcf, 0x05, 0x7a, 0x0c, 0xbd, 0x85, 0x64, 0x54, 0xb3, 0x29, 0x95, 0x3a, 0xd5,
-	0x29, 0xcf, 0x26, 0x22, 0xf2, 0xfb, 0xde, 0x71, 0xe7, 0x34, 0x72, 0x9e, 0x67, 0xcd, 0x7d, 0xb2,
-	0x79, 0x04, 0x7d, 0x0e, 0x1d, 0xb5, 0x92, 0x69, 0xf6, 0xd3, 0x70, 0x46, 0x26, 0x22, 0x6a, 0xd9,
-	0x08, 0xc8, 0x45, 0x98, 0x55, 0x3b, 0xa4, 0xee, 0x86, 0xbe, 0x82, 0xee, 0x62, 0x45, 0xb3, 0x25,
-	0x1b, 0x31, 0x9a, 0x30, 0x39, 0x11, 0xd1, 0x8e, 0x3d, 0xf8, 0x4e, 0x91, 0x7a, 0x6d, 0x93, 0x34,
-	0x9c, 0x4d, 0x52, 0xf6, 0x52, 0xd0, 0x2c, 0xc9, 0x93, 0xb6, 0xd7, 0x92, 0xc6, 0xd5, 0x0e, 0xa9,
-	0xbb, 0xe1, 0x33, 0xe8, 0x6d, 0x94, 0x84, 0x4e, 0x20, 0x10, 0x05, 0xb4, 0x4a, 0x75, 0x4e, 0x43,
-	0x17, 0xa8, 0x74, 0x23, 0x95, 0x0b, 0xfe, 0xc3, 0x83, 0x4e, 0xad, 0x2c, 0x74, 0x00, 0xbb, 0x4a,
-	0x4b, 0x46, 0xaf, 0x5d, 0x23, 0x1c, 0x42, 0x87, 0xf5, 0xb8, 0x46, 0xd7, 0x76, 0x2d, 0x0a, 0x3a,
-	0x86, 0xfb, 0x92, 0x89, 0xab, 0x74, 0x41, 0xe7, 0x9c, 0xb0, 0x6b, 0xfe, 0x82, 0x59, 0xe5, 0x02,
-	0xd2, 0x34, 0x9b, 0xf8, 0x57, 0xb6, 0x6c, 0xab, 0x50, 0x40, 0x1c, 0x42, 0x7d, 0xe8, 0xe4, 0xab,
-	0x58, 0xf0, 0xc5, 0xca, 0x4a, 0xb0, 0x43, 0xea, 0x26, 0xfc, 0xca, 0x83, 0x4e, 0x4d, 0x8b, 0x2d,
-	0x99, 0x62, 0xb8, 0x5b, 0x52, 0x1a, 0x24, 0x89, 0xa3, 0xb9, 0x66, 0x7b, 0x03, 0x8e, 0xbf, 0x7b,
-	0xd0, 0x25, 0x4c, 0x70, 0xa9, 0xcb, 0xde, 0x6e, 0x47, 0x33, 0x82, 0x3d, 0x47, 0xc9, 0x31, 0x2c,
-	0xe0, 0x1b, 0x90, 0xfb, 0x1e, 0xba, 0xeb, 0xf7, 0x70, 0x4b, 0x6e, 0x15, 0x83, 0x56, 0x9d, 0x01,
-	0xfe, 0xc5, 0x87, 0x60, 0x5a, 0xaf, 0x40, 0xdd, 0x3c, 0xff, 0x91, 0x2d, 0xb4, 0x0b, 0x5e, 0xc0,
-	0x5a, 0x56, 0x7f, 0x2d, 0x6b, 0x17, 0xfc, 0x34, 0x6f, 0x48, 0x9b, 0xf8, 0x69, 0x82, 0x1e, 0x40,
-	0x7b, 0x29, 0xf9, 0x8d, 0x70, 0x85, 0xe6, 0x00, 0x7d, 0x0c, 0x3d, 0x27, 0x85, 0x49, 0xf3, 0x98,
-	0x2e, 0x34, 0x97, 0xb6, 0xda, 0x36, 0xd9, 0xdc, 0x30, 0x93, 0xc5, 0x19, 0x55, 0xb4, 0xdb, 0x6f,
-	0x99, 0xc9, 0x52, 0xe0, 0x5a, 0x1d, 0x7b, 0x6b, 0x4a, 0x86, 0xd0, 0x4a, 0x95, 0x8c, 0xf6, 0xad,
-	0xbb, 0x59, 0x36, 0xb5, 0x0d, 0x36, 0xb4, 0x35, 0x5c, 0x99, 0xdd, 0x03, 0xbb, 0x97, 0x03, 0x1c,
-	0xc3, 0x7d, 0x33, 0xba, 0x9e, 0xf2, 0x34, 0x23, 0xec, 0xe7, 0x1b, 0xa6, 0x6c, 0xf1, 0x19, 0x4f,
-	0x58, 0x39, 0xe8, 0x1c, 0x32, 0x44, 0xcd, 0x6a, 0x90, 0x24, 0xd2, 0xc9, 0x52, 0x62, 0x7c, 0x0c,
-	0x61, 0x15, 0x46, 0x09, 0x9e, 0x29, 0x66, 0x13, 0x4a, 0xc9, 0xa5, 0x0b, 0x93, 0x03, 0x7c, 0x0e,
-	0xe1, 0x33, 0xa6, 0x69, 0x42, 0x35, 0x9d, 0x65, 0x54, 0xa8, 0x15, 0xd7, 0xe8, 0x53, 0x80, 0xb2,
-	0x77, 0x2a, 0xf2, 0xfa, 0xad, 0xd7, 0x8e, 0x84, 0x9a, 0x0f, 0x7e, 0x0a, 0x88, 0x54, 0x4a, 0x16,
-	0xcc, 0x0f, 0x21, 0x70, 0xd2, 0x95, 0xe4, 0x2b, 0x83, 0xa9, 0x8b, 0x5f, 0x5e, 0x2a, 0xa6, 0x2d,
-	0xfb, 0x16, 0x71, 0x08, 0x7f, 0x09, 0xd1, 0xa8, 0xd2, 0x69, 0x62, 0x8d, 0x45, 0xc4, 0x86, 0xac,
-	0xde, 0xe6, 0x95, 0xfd, 0x02, 0x1e, 0xbe, 0xe6, 0xb4, 0x93, 0xe0, 0x10, 0x02, 0x96, 0x25, 0xb9,
-	0xd1, 0x1e, 0x6e, 0x91, 0xca, 0x80, 0x5f, 0xf9, 0xd0, 0x9b, 0x4a, 0x2e, 0xe8, 0x92, 0x6a, 0x96,
-	0x14, 0x29, 0xdf, 0xe6, 0x17, 0x44, 0xae, 0x0d, 0x8e, 0xc6, 0x0b, 0xb2, 0x3e, 0x55, 0x48, 0xc3,
-	0x79, 0xcb, 0x17, 0xe4, 0x13, 0x68, 0xc7, 0xe6, 0xde, 0x20, 0x04, 0x3b, 0x0b, 0x9e, 0x30, 0x2b,
-	0xcc, 0x3d, 0x62, 0xd7, 0xe6, 0x33, 0xb8, 0x56, 0x4b, 0x77, 0x19, 0xcd, 0x12, 0xcf, 0x00, 0xd5,
-	0x15, 0x75, 0x6d, 0xf8, 0x0f, 0x49, 0x71, 0x71, 0x49, 0x73, 0x19, 0xef, 0x16, 0x7c, 0x8c, 0xad,
-	0xb8, 0xb2, 0x1f, 0x40, 0x2f, 0xff, 0x2b, 0x30, 0xcc, 0x2e, 0x79, 0xd1, 0xa6, 0x7c, 0x14, 0xe4,
-	0x97, 0xcc, 0x4f, 0x13, 0x3c, 0x02, 0x54, 0x77, 0x72, 0x99, 0x1b, 0x5e, 0xa6, 0x8a, 0x15, 0x57,
-	0xda, 0x51, 0xb6, 0x6b, 0x63, 0x33, 0x42, 0xb9, 0xb1, 0x62, 0xd7, 0x78, 0x0c, 0x07, 0x65, 0xc3,
-	0xcc, 0x1f, 0x90, 0x1b, 0x55, 0xfb, 0x3a, 0xff, 0xff, 0x40, 0xc4, 0xcf, 0xe0, 0xdd, 0x8d, 0x78,
-	0x8e, 0xe2, 0x01, 0xec, 0xb2, 0x97, 0xa9, 0xd2, 0xca, 0x06, 0xdc, 0x27, 0x0e, 0x99, 0xcf, 0x3d,
-	0x55, 0x79, 0xf7, 0x6c, 0xbc, 0x7d, 0x52, 0xe2, 0x8f, 0x7e, 0x00, 0x7f, 0x22, 0xd0, 0x03, 0x08,
-	0xcf, 0x48, 0x3c, 0x98, 0xc7, 0x17, 0xd3, 0x01, 0x99, 0x0f, 0xe7, 0xc3, 0xc9, 0x38, 0xbc, 0x83,
-	0xba, 0x00, 0xb3, 0x27, 0x64, 0x38, 0xfe, 0xf6, 0x62, 0x38, 0x23, 0xa1, 0x87, 0x7a, 0x70, 0x8f,
-	0xc4, 0xd3, 0x09, 0x99, 0x5f, 0x8c, 0xe2, 0xc1, 0x79, 0x4c, 0x42, 0xdf, 0x98, 0xce, 0x9e, 0x0c,
-	0xc6, 0xdf, 0xc4, 0x85, 0xa9, 0x65, 0x4e, 0xc5, 0xdf, 0x4d, 0x07, 0xe3, 0x73, 0x7b, 0x6a, 0xe7,
-	0xeb, 0xf0, 0xcf, 0xdb, 0x23, 0xef, 0xaf, 0xdb, 0x23, 0xef, 0xef, 0xdb, 0x23, 0xef, 0xb7, 0x7f,
-	0x8e, 0xee, 0x3c, 0xdf, 0xb5, 0x9d, 0xf9, 0xec, 0xdf, 0x00, 0x00, 0x00, 0xff, 0xff, 0x2c, 0x53,
-	0x25, 0x44, 0xbd, 0x09, 0x00, 0x00,
+	0x14, 0xef, 0xae, 0xe3, 0x24, 0xfb, 0xdc, 0xba, 0x6b, 0x4d, 0x1b, 0xb6, 0x90, 0xc9, 0x78, 0xc4,
+	0x25, 0x30, 0x90, 0x32, 0x81, 0x0b, 0x33, 0x70, 0x30, 0xc9, 0x96, 0xba, 0x38, 0xb6, 0x47, 0xf6,
+	0x81, 0x13, 0x41, 0xf5, 0xca, 0xf6, 0x42, 0xb2, 0x12, 0x92, 0xd2, 0xe9, 0xb7, 0xe0, 0x08, 0xc3,
+	0xad, 0x27, 0xbe, 0x0a, 0x47, 0x3e, 0x02, 0x13, 0xbe, 0x08, 0x23, 0xad, 0xf6, 0x8f, 0xed, 0x0e,
+	0x33, 0x38, 0x17, 0x4e, 0xd6, 0xef, 0xe9, 0xe9, 0xbd, 0xdf, 0xfb, 0x3d, 0xe9, 0xad, 0xe1, 0x3d,
+	0xc5, 0xe4, 0x2b, 0x26, 0x9f, 0x0a, 0xc9, 0x35, 0x7f, 0x9a, 0x66, 0x9a, 0xc9, 0x8c, 0x5e, 0x9d,
+	0x58, 0x88, 0x9a, 0xf6, 0x07, 0x7f, 0x00, 0xad, 0x89, 0xf5, 0x9a, 0x68, 0xaa, 0x19, 0x7a, 0x17,
+	0xf6, 0xf3, 0x43, 0xfd, 0xf3, 0xc8, 0xeb, 0x7a, 0xc7, 0x01, 0x29, 0x31, 0xfe, 0xc5, 0x87, 0x3d,
+	0x42, 0xe7, 0x7a, 0xc0, 0x17, 0xe8, 0x09, 0xf8, 0x5c, 0x58, 0x8f, 0xf6, 0x69, 0x90, 0x47, 0x3c,
+	0x19, 0x09, 0xe2, 0x73, 0x81, 0x9e, 0x41, 0x67, 0x26, 0x19, 0xd5, 0x6c, 0x4c, 0xa5, 0x4e, 0x75,
+	0xca, 0xb3, 0x91, 0x88, 0xfc, 0xae, 0x77, 0xdc, 0x3a, 0x8d, 0x9c, 0xe7, 0xd9, 0xfa, 0x3e, 0xd9,
+	0x3c, 0x82, 0x3e, 0x83, 0x96, 0x5a, 0xca, 0x34, 0xfb, 0xb1, 0x3f, 0x21, 0x23, 0x11, 0x35, 0x6c,
+	0x04, 0xe4, 0x22, 0x4c, 0xaa, 0x1d, 0x52, 0x77, 0x43, 0x5f, 0x42, 0x7b, 0xb6, 0xa4, 0xd9, 0x82,
+	0x0d, 0x18, 0x4d, 0x98, 0x1c, 0x89, 0x68, 0xc7, 0x1e, 0x7c, 0x5c, 0xa4, 0x5e, 0xd9, 0x24, 0x6b,
+	0xce, 0x26, 0x29, 0x7b, 0x2d, 0x68, 0x96, 0xe4, 0x49, 0x9b, 0x2b, 0x49, 0xe3, 0x6a, 0x87, 0xd4,
+	0xdd, 0xf0, 0x19, 0x74, 0x36, 0x4a, 0x42, 0x27, 0x10, 0x88, 0x02, 0x5a, 0xa5, 0x5a, 0xa7, 0xa1,
+	0x0b, 0x54, 0xba, 0x91, 0xca, 0x05, 0xff, 0xee, 0x41, 0xab, 0x56, 0x16, 0x3a, 0x80, 0x5d, 0xa5,
+	0x25, 0xa3, 0xd7, 0xae, 0x11, 0x0e, 0xa1, 0xc3, 0x7a, 0x5c, 0xa3, 0x6b, 0xb3, 0x16, 0x05, 0x1d,
+	0xc3, 0x43, 0xc9, 0xc4, 0x55, 0x3a, 0xa3, 0x53, 0x4e, 0xd8, 0x35, 0x7f, 0xc5, 0xac, 0x72, 0x01,
+	0x59, 0x37, 0x9b, 0xf8, 0x57, 0xb6, 0x6c, 0xab, 0x50, 0x40, 0x1c, 0x42, 0x5d, 0x68, 0xe5, 0xab,
+	0x58, 0xf0, 0xd9, 0xd2, 0x4a, 0xb0, 0x43, 0xea, 0x26, 0xfc, 0xc6, 0x83, 0x56, 0x4d, 0x8b, 0x2d,
+	0x99, 0x62, 0xb8, 0x5f, 0x52, 0xea, 0x25, 0x89, 0xa3, 0xb9, 0x62, 0xbb, 0x03, 0xc7, 0xdf, 0x3c,
+	0x68, 0x13, 0x26, 0xb8, 0xd4, 0x65, 0x6f, 0xb7, 0xa3, 0x19, 0xc1, 0x9e, 0xa3, 0xe4, 0x18, 0x16,
+	0xf0, 0x0e, 0xe4, 0xbe, 0x83, 0xf6, 0xea, 0x3d, 0xdc, 0x92, 0x5b, 0xc5, 0xa0, 0x51, 0x67, 0x80,
+	0x7f, 0xf6, 0x21, 0x18, 0xd7, 0x2b, 0x50, 0x37, 0x2f, 0x7f, 0x60, 0x33, 0xed, 0x82, 0x17, 0xb0,
+	0x96, 0xd5, 0x5f, 0xc9, 0xda, 0x06, 0x3f, 0xcd, 0x1b, 0xd2, 0x24, 0x7e, 0x9a, 0xa0, 0x47, 0xd0,
+	0x5c, 0x48, 0x7e, 0x23, 0x5c, 0xa1, 0x39, 0x40, 0x1f, 0x41, 0xc7, 0x49, 0x61, 0xd2, 0x3c, 0xa3,
+	0x33, 0xcd, 0xa5, 0xad, 0xb6, 0x49, 0x36, 0x37, 0xcc, 0x64, 0x71, 0x46, 0x15, 0xed, 0x76, 0x1b,
+	0x66, 0xb2, 0x14, 0xb8, 0x56, 0xc7, 0xde, 0x8a, 0x92, 0x21, 0x34, 0x52, 0x25, 0xa3, 0x7d, 0xeb,
+	0x6e, 0x96, 0xeb, 0xda, 0x06, 0x1b, 0xda, 0x1a, 0xae, 0xcc, 0xee, 0x81, 0xdd, 0xcb, 0x01, 0x8e,
+	0xe1, 0xa1, 0x19, 0x5d, 0x2f, 0x78, 0x9a, 0x11, 0xf6, 0xd3, 0x0d, 0x53, 0xb6, 0xf8, 0x8c, 0x27,
+	0xac, 0x1c, 0x74, 0x0e, 0x19, 0xa2, 0x66, 0xd5, 0x4b, 0x12, 0xe9, 0x64, 0x29, 0x31, 0x3e, 0x86,
+	0xb0, 0x0a, 0xa3, 0x04, 0xcf, 0x14, 0xb3, 0x09, 0xa5, 0xe4, 0xd2, 0x85, 0xc9, 0x01, 0x3e, 0x87,
+	0xf0, 0x82, 0x69, 0x9a, 0x50, 0x4d, 0x27, 0x19, 0x15, 0x6a, 0xc9, 0x35, 0xfa, 0x04, 0xa0, 0xec,
+	0x9d, 0x8a, 0xbc, 0x6e, 0xe3, 0xad, 0x23, 0xa1, 0xe6, 0x83, 0x5f, 0x00, 0x22, 0x95, 0x92, 0x05,
+	0xf3, 0x43, 0x08, 0x9c, 0x74, 0x25, 0xf9, 0xca, 0x60, 0xea, 0xe2, 0xf3, 0xb9, 0x62, 0xda, 0xb2,
+	0x6f, 0x10, 0x87, 0xf0, 0x17, 0x10, 0x0d, 0x2a, 0x9d, 0x46, 0xd6, 0x58, 0x44, 0x5c, 0x93, 0xd5,
+	0xdb, 0xbc, 0xb2, 0x9f, 0xc3, 0x93, 0xb7, 0x9c, 0x76, 0x12, 0x1c, 0x42, 0xc0, 0xb2, 0x24, 0x37,
+	0xda, 0xc3, 0x0d, 0x52, 0x19, 0xf0, 0x1b, 0x1f, 0x3a, 0x63, 0xc9, 0x05, 0x5d, 0x50, 0xcd, 0x92,
+	0x22, 0xe5, 0xff, 0xf9, 0x0b, 0x22, 0x57, 0x06, 0xc7, 0xda, 0x17, 0x64, 0x75, 0xaa, 0x90, 0x35,
+	0xe7, 0x2d, 0xbf, 0x20, 0x1f, 0x43, 0x33, 0x36, 0xf7, 0x06, 0x21, 0xd8, 0x99, 0xf1, 0x84, 0x59,
+	0x61, 0x1e, 0x10, 0xbb, 0x36, 0xcf, 0xe0, 0x5a, 0x2d, 0xdc, 0x65, 0x34, 0x4b, 0x3c, 0x01, 0x54,
+	0x57, 0xd4, 0xb5, 0xe1, 0x5f, 0x24, 0xc5, 0xc5, 0x25, 0xcd, 0x65, 0xbc, 0x5f, 0xf0, 0x31, 0xb6,
+	0xe2, 0xca, 0xbe, 0x0f, 0x9d, 0xfc, 0xaf, 0x40, 0x3f, 0x9b, 0xf3, 0xa2, 0x4d, 0xf9, 0x28, 0xc8,
+	0x2f, 0x99, 0x9f, 0x26, 0x78, 0x00, 0xa8, 0xee, 0xe4, 0x32, 0xaf, 0x79, 0x99, 0x2a, 0x96, 0x5c,
+	0x69, 0x47, 0xd9, 0xae, 0x8d, 0xcd, 0x08, 0xe5, 0xc6, 0x8a, 0x5d, 0xe3, 0x21, 0x1c, 0x94, 0x0d,
+	0x33, 0x7f, 0x40, 0x6e, 0x54, 0xed, 0x75, 0xfe, 0xf7, 0x81, 0x88, 0x2f, 0xe0, 0x9d, 0x8d, 0x78,
+	0x8e, 0xe2, 0x01, 0xec, 0xb2, 0xd7, 0xa9, 0xd2, 0xca, 0x06, 0xdc, 0x27, 0x0e, 0x99, 0xe7, 0x9e,
+	0xaa, 0xbc, 0x7b, 0x36, 0xde, 0x3e, 0x29, 0x31, 0xbe, 0x80, 0xc7, 0x65, 0xb8, 0x21, 0xd7, 0xe9,
+	0xdc, 0x3d, 0xc4, 0xed, 0xd8, 0x7d, 0xf8, 0x3d, 0xf8, 0x23, 0x81, 0x1e, 0x41, 0x78, 0x46, 0xe2,
+	0xde, 0x34, 0xbe, 0x1c, 0xf7, 0xc8, 0xb4, 0x3f, 0xed, 0x8f, 0x86, 0xe1, 0x3d, 0xd4, 0x06, 0x98,
+	0x3c, 0x27, 0xfd, 0xe1, 0x37, 0x97, 0xfd, 0x09, 0x09, 0x3d, 0xd4, 0x81, 0x07, 0x24, 0x1e, 0x8f,
+	0xc8, 0xf4, 0x72, 0x10, 0xf7, 0xce, 0x63, 0x12, 0xfa, 0xc6, 0x74, 0xf6, 0xbc, 0x37, 0xfc, 0x3a,
+	0x2e, 0x4c, 0x0d, 0x73, 0x2a, 0xfe, 0x76, 0xdc, 0x1b, 0x9e, 0xdb, 0x53, 0x3b, 0x5f, 0x85, 0x7f,
+	0xdc, 0x1e, 0x79, 0x7f, 0xde, 0x1e, 0x79, 0x7f, 0xdd, 0x1e, 0x79, 0xbf, 0xfe, 0x7d, 0x74, 0xef,
+	0xe5, 0xae, 0x6d, 0xf4, 0xa7, 0xff, 0x04, 0x00, 0x00, 0xff, 0xff, 0xe7, 0xa9, 0x08, 0x9a, 0x0c,
+	0x0a, 0x00, 0x00,
 }
diff --git a/server/proto/internal.proto b/server/proto/internal.proto
index 018a9cc..e8ce0f4 100644
--- a/server/proto/internal.proto
+++ b/server/proto/internal.proto
@@ -138,3 +138,8 @@ message PartitionStatusResponse {
     bool exists   = 1;
     bool isLeader = 2;
 }
+
+message PartitionNotification {
+    string stream    = 1;
+    int32  partition = 2;
+}
diff --git a/server/replicator.go b/server/replicator.go
index 7bccf2e..6be3eb2 100644
--- a/server/replicator.go
+++ b/server/replicator.go
@@ -28,7 +28,8 @@ const (
 // where responses should be sent.
 type replicationRequest struct {
 	*proto.ReplicationRequest
-	request *nats.Msg
+	request  *nats.Msg
+	received time.Time
 }
 
 // replicator handles replication requests from a particular replica and tracks
@@ -47,6 +48,19 @@ type replicator struct {
 	leader       string
 	epoch        uint64
 	headersBuf   [28]byte // scratch buffer for reading message headers
+	writer       replicationProtocolWriter
+	waiter       <-chan struct{}
+}
+
+func newReplicator(epoch uint64, replica string, p *partition) *replicator {
+	return &replicator{
+		epoch:      epoch,
+		replica:    replica,
+		partition:  p,
+		requests:   make(chan replicationRequest, 1),
+		maxLagTime: p.srv.config.Clustering.ReplicaMaxLagTime,
+		leader:     p.srv.config.Clustering.ServerID,
+	}
 }
 
 // start a long-running replication loop for the given leader epoch until the
@@ -56,12 +70,12 @@ type replicator struct {
 // available. The response will also include the leader epoch and HW. If the
 // replica doesn't send a request or catch up to the leader's log in
 // maxLagTime, it will be removed from the ISR until it catches back up.
-func (r *replicator) start(epoch uint64, stop chan struct{}) {
+func (r *replicator) start(stop <-chan struct{}) {
 	r.mu.Lock()
-	r.epoch = epoch
 	now := time.Now()
 	r.lastSeen = now
 	r.lastCaughtUp = now
+	r.writer = newReplicationProtocolWriter(r, stop)
 	r.mu.Unlock()
 
 	// Start a goroutine to track the replica's health.
@@ -75,9 +89,8 @@ func (r *replicator) start(epoch uint64, stop chan struct{}) {
 		case req = <-r.requests:
 		}
 
-		now := time.Now()
 		r.mu.Lock()
-		r.lastSeen = now
+		r.lastSeen = req.received
 		r.mu.Unlock()
 
 		// Update the ISR replica's latest offset for the partition. This is
@@ -91,10 +104,7 @@ func (r *replicator) start(epoch uint64, stop chan struct{}) {
 
 		// Check if we're caught up.
 		if req.Offset >= latest {
-			r.mu.Lock()
-			r.lastCaughtUp = now
-			r.mu.Unlock()
-			r.sendHW(req.request)
+			r.caughtUp(stop, latest, req)
 			continue
 		}
 
@@ -109,14 +119,20 @@ func (r *replicator) start(epoch uint64, stop chan struct{}) {
 					"and replica %s (requested offset %d, earliest %d, latest %d): %v",
 				r.partition, r.replica, req.Offset+1, earliest, latest, err)
 			// Send a response to short-circuit request timeout.
-			r.sendHW(req.request)
+			if err := r.sendHW(req.request); err != nil {
+				r.partition.srv.logger.Errorf("Failed to send HW for partition %s to replica %s: %v",
+					r.partition, req.ReplicaID, err)
+			}
 			continue
 		}
 
 		// Send a batch of messages to the replica.
 		if err := r.replicate(ctx, reader, req.request, req.Offset); err != nil {
 			// Send a response to short-circuit request timeout.
-			r.sendHW(req.request)
+			if err := r.sendHW(req.request); err != nil {
+				r.partition.srv.logger.Errorf("Failed to send HW for partition %s to replica %s: %v",
+					r.partition, req.ReplicaID, err)
+			}
 			continue
 		}
 	}
@@ -135,7 +151,7 @@ func (r *replicator) request(req replicationRequest) {
 // any replication requests or hasn't consumed up to the leader's log end
 // offset for the lag-time duration. If this is the case, the follower is
 // removed from the ISR until it catches back up.
-func (r *replicator) tick(stop chan struct{}) {
+func (r *replicator) tick(stop <-chan struct{}) {
 	ticker := time.NewTicker(r.maxLagTime)
 	defer ticker.Stop()
 	var now time.Time
@@ -208,25 +224,12 @@ func (r *replicator) expandISR() {
 func (r *replicator) replicate(
 	ctx context.Context, reader *commitlog.Reader, request *nats.Msg, offset int64) error {
 
-	buf := new(bytes.Buffer)
-	// Write the leader epoch to the buffer.
-	if err := binary.Write(buf, proto.Encoding, r.epoch); err != nil {
-		r.partition.srv.logger.Errorf("Failed to write leader epoch to buffer while replicating: %v", err)
-		return err
-	}
-	// Reserve space for the HW. This will be replaced with the HW at the time
-	// of flush.
-	if err := binary.Write(buf, proto.Encoding, int64(0)); err != nil {
-		r.partition.srv.logger.Errorf("Failed to write HW to buffer while replicating: %v", err)
-		return err
-	}
-
 	var (
 		newestOffset = r.partition.log.NewestOffset()
 		message      commitlog.Message
 		err          error
 	)
-	for offset < newestOffset && buf.Len() < replicationMaxSize {
+	for offset < newestOffset && r.writer.Len() < replicationMaxSize {
 		message, offset, _, _, err = reader.ReadMessage(ctx, r.headersBuf[:])
 		if err != nil {
 			r.partition.srv.logger.Errorf("Failed to read message while replicating: %v", err)
@@ -235,39 +238,124 @@ func (r *replicator) replicate(
 
 		// Check if this message will put us over the batch size limit. If it
 		// does, flush the batch now.
-		if uint32(len(message))+uint32(len(r.headersBuf))+uint32(buf.Len()) > replicationMaxSize {
+		if uint32(len(message))+uint32(len(r.headersBuf))+uint32(r.writer.Len()) > replicationMaxSize {
 			break
 		}
 
 		// Write the message to the buffer.
-		if err := writeMessageToBuffer(buf, r.headersBuf[:], message); err != nil {
+		if err := r.writer.Write(offset, r.headersBuf[:], message); err != nil {
 			r.partition.srv.logger.Errorf("Failed to write message to buffer while replicating: %v", err)
 			return err
 		}
 	}
 
-	// Set the HW and flush the batch.
-	data := buf.Bytes()
-	proto.Encoding.PutUint64(data[8:], uint64(r.partition.log.HighWatermark()))
-	return request.Respond(data)
+	// Flush the batch.
+	if err := r.writer.Flush(request.Respond); err != nil {
+		r.partition.srv.logger.Errorf("Failed to flush buffer while replicating: %v", err)
+		return err
+	}
+	return nil
+}
+
+// caughtUp is called when the follower has caught up with the leader's log.
+// This will register a data waiter on the log so that the leader can notify
+// the follower when new data is available to replicate.
+func (r *replicator) caughtUp(stop <-chan struct{}, leo int64, req replicationRequest) {
+	r.mu.Lock()
+	r.lastCaughtUp = req.received
+	waiter := r.waiter
+	if waiter == nil {
+		// Register a waiter to be notified when new messages are written after
+		// the current log end offset to preempt an idle follower.
+		waiter = r.partition.log.NotifyLEO(r, leo)
+		r.partition.srv.startGoroutine(func() {
+			select {
+			case <-waiter:
+				r.partition.sendPartitionNotification(req.ReplicaID)
+				r.mu.Lock()
+				r.waiter = nil
+				r.mu.Unlock()
+			case <-stop:
+			}
+		})
+		r.waiter = waiter
+	}
+	r.mu.Unlock()
+
+	if err := r.sendHW(req.request); err != nil {
+		r.partition.srv.logger.Errorf("Failed to send HW for partition %s to replica %s: %v",
+			r.partition, req.ReplicaID, err)
+	}
 }
 
-// writeMessageToBuffer writes the headers and message byte slices to the bytes
-// buffer.
-func writeMessageToBuffer(buf *bytes.Buffer, headers, message []byte) error {
-	if _, err := buf.Write(headers); err != nil {
+// sendHW sends the leader epoch and HW to the given NATS inbox.
+func (r *replicator) sendHW(request *nats.Msg) error {
+	r.writer.Reset()
+	return r.writer.Flush(request.Respond)
+}
+
+type replicationProtocolWriter interface {
+	Write(offset int64, headers, message []byte) error
+	Flush(func(data []byte) error) error
+	Len() int
+	Reset()
+}
+
+type protocolWriter struct {
+	*replicator
+	buf        *bytes.Buffer
+	log        CommitLog
+	lastOffset int64
+	stop       <-chan struct{}
+}
+
+func newReplicationProtocolWriter(r *replicator, stop <-chan struct{}) replicationProtocolWriter {
+	w := &protocolWriter{
+		replicator: r,
+		buf:        new(bytes.Buffer),
+		log:        r.partition.log,
+		stop:       stop,
+	}
+	w.Reset()
+	return w
+}
+
+func (w *protocolWriter) Write(offset int64, headers, message []byte) error {
+	if _, err := w.buf.Write(headers); err != nil {
 		return err
 	}
-	if _, err := buf.Write(message); err != nil {
+	if _, err := w.buf.Write(message); err != nil {
 		return err
 	}
+	w.lastOffset = offset
 	return nil
 }
 
-// sendHW sends the leader epoch and HW to the given NATS inbox.
-func (r *replicator) sendHW(request *nats.Msg) {
-	buf := make([]byte, replicationOverhead)
-	proto.Encoding.PutUint64(buf[:8], r.epoch)
-	proto.Encoding.PutUint64(buf[8:], uint64(r.partition.log.HighWatermark()))
-	request.Respond(buf)
+func (w *protocolWriter) Flush(write func([]byte) error) error {
+	data := w.buf.Bytes()
+	// Replace the HW.
+	proto.Encoding.PutUint64(data[8:], uint64(w.log.HighWatermark()))
+
+	if err := write(data); err != nil {
+		w.Reset()
+		return err
+	}
+
+	w.Reset()
+	return nil
+}
+
+func (w *protocolWriter) Len() int {
+	return w.buf.Len()
+}
+
+func (w *protocolWriter) Reset() {
+	w.buf.Reset()
+	w.lastOffset = -1
+
+	// Write the leader epoch.
+	binary.Write(w.buf, proto.Encoding, w.replicator.epoch)
+	// Reserve space for the HW. This will be replaced with the HW at the time
+	// of flush.
+	binary.Write(w.buf, proto.Encoding, int64(0))
 }
diff --git a/server/server.go b/server/server.go
index 754322d..f054a7b 100644
--- a/server/server.go
+++ b/server/server.go
@@ -27,11 +27,7 @@ import (
 	"github.com/liftbridge-io/liftbridge/server/proto"
 )
 
-const (
-	stateFile                    = "liftbridge"
-	serverInfoInboxTemplate      = "%s.info"
-	partitionStatusInboxTemplate = "%s.status.%s"
-)
+const stateFile = "liftbridge"
 
 // Server is the main Liftbridge object. Create it by calling New or
 // RunServerWithConfig.
@@ -149,14 +145,20 @@ func (s *Server) Start() (err error) {
 		return errors.Wrap(err, "failed to start Raft node")
 	}
 
-	if _, err := s.ncRaft.Subscribe(s.serverInfoInbox(), s.handleServerInfoRequest); err != nil {
+	if _, err := s.ncRaft.Subscribe(s.getServerInfoInbox(), s.handleServerInfoRequest); err != nil {
 		return errors.Wrap(err, "failed to subscribe to server info subject")
 	}
 
-	if _, err := s.ncRaft.Subscribe(s.partitionStatusInbox(), s.handlePartitionStatusRequest); err != nil {
+	inbox := s.getPartitionStatusInbox(s.config.Clustering.ServerID)
+	if _, err := s.ncRaft.Subscribe(inbox, s.handlePartitionStatusRequest); err != nil {
 		return errors.Wrap(err, "failed to subscribe to partition status subject")
 	}
 
+	inbox = s.getPartitionNotificationInbox(s.config.Clustering.ServerID)
+	if _, err := s.ncRepl.Subscribe(inbox, s.handlePartitionNotification); err != nil {
+		return errors.Wrap(err, "failed to subscribe to partition notification subject")
+	}
+
 	s.handleSignals()
 
 	return errors.Wrap(s.startAPIServer(), "failed to start API server")
@@ -657,7 +659,7 @@ func (s *Server) handleServerInfoRequest(m *nats.Msg) {
 func (s *Server) handlePartitionStatusRequest(m *nats.Msg) {
 	req := &proto.PartitionStatusRequest{}
 	if err := req.Unmarshal(m.Data); err != nil {
-		s.logger.Warn("Dropping invalid partition status request: %v", err)
+		s.logger.Warnf("Dropping invalid partition status request: %v", err)
 		return
 	}
 
@@ -678,17 +680,51 @@ func (s *Server) handlePartitionStatusRequest(m *nats.Msg) {
 	}
 }
 
-// serverInfoInbox returns the NATS subject used for handling server
+// handlePartitionNotification is a NATS handler used to process notifications
+// from a leader that new data is available on a partition for the follower to
+// replicate if the follower is idle.
+//
+// When a follower reaches the end of the log, it starts to sleep in between
+// replication requests to avoid overloading the leader. However, this causes
+// added commit latency when new messages are published to the log since the
+// follower is idle. As a result, the leader will note when a follower is
+// caught up and send a notification in order to wake an idle follower back up
+// when new data is written to the log.
+func (s *Server) handlePartitionNotification(m *nats.Msg) {
+	req := &proto.PartitionNotification{}
+	if err := req.Unmarshal(m.Data); err != nil {
+		s.logger.Warnf("Dropping invalid partition notification: %v", err)
+		return
+	}
+
+	partition := s.metadata.GetPartition(req.Stream, req.Partition)
+	if partition == nil {
+		s.logger.Warnf("Dropping invalid partition notification: no partition %d for stream %s",
+			req.Partition, req.Stream)
+		return
+	}
+
+	// Wake the follower up.
+	partition.Notify()
+}
+
+// getServerInfoInbox returns the NATS subject used for handling server
 // information requests.
-func (s *Server) serverInfoInbox() string {
-	return fmt.Sprintf(serverInfoInboxTemplate, s.baseMetadataRaftSubject())
+func (s *Server) getServerInfoInbox() string {
+	return fmt.Sprintf("%s.info", s.baseMetadataRaftSubject())
+}
+
+// getPartitionStatusInbox returns the NATS subject used for handling stream
+// status requests.
+func (s *Server) getPartitionStatusInbox(id string) string {
+	return fmt.Sprintf("%s.status.%s", s.baseMetadataRaftSubject(), id)
 }
 
-// partitionStatusInbox returns the NATS subject used for handling stream status
-// requests.
-func (s *Server) partitionStatusInbox() string {
-	return fmt.Sprintf(partitionStatusInboxTemplate, s.baseMetadataRaftSubject(),
-		s.config.Clustering.ServerID)
+// getPartitionNotificationInbox returns the NATS subject used for leaders to
+// indicate new data is available on a partition for a follower to replicate if
+// the follower is idle.
+func (s *Server) getPartitionNotificationInbox(id string) string {
+	return fmt.Sprintf("%s.notify.%s", s.config.Clustering.Namespace, id)
 }
 
 // startGoroutine starts a goroutine which is managed by the server. This adds


Test patch:
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 496d977..e591464 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -433,6 +433,114 @@ func TestTruncate(t *testing.T) {
 	require.Equal(t, int64(5), l.LastOffsetForLeaderEpoch(1))
 }
 
+// Ensure NotifyLEO returns a closed channel when the given offset is not the
+// current log end offset.
+func TestNotifyLEOMismatch(t *testing.T) {
+	l, cleanup := setup(t)
+	defer l.Close()
+	defer cleanup()
+
+	// Add some messages.
+	for i := 0; i < 5; i++ {
+		_, err := l.Append([]*proto.Message{&proto.Message{
+			Value:       []byte(strconv.Itoa(i)),
+			Timestamp:   time.Now().UnixNano(),
+			LeaderEpoch: 1,
+		}})
+		require.NoError(t, err)
+	}
+
+	// Get current log end offset and then add another message.
+	leo := l.NewestOffset()
+	_, err := l.Append([]*proto.Message{&proto.Message{
+		Value:       []byte(strconv.Itoa(5)),
+		Timestamp:   time.Now().UnixNano(),
+		LeaderEpoch: 1,
+	}})
+	require.NoError(t, err)
+
+	// Notify LEO should return a closed channel because the LEO is different
+	// than the expected LEO.
+	waiter := struct{}{}
+	ch := l.NotifyLEO(waiter, leo)
+	select {
+	case <-ch:
+	default:
+		t.Fatalf("Expected closed channel")
+	}
+}
+
+// Ensure NotifyLEO returns a channel that is closed once more data is written
+// to the log past the log end offset.
+func TestNotifyLEONewData(t *testing.T) {
+	l, cleanup := setupWithOptions(t, Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 256,
+	})
+	defer l.Close()
+	defer cleanup()
+
+	// Add some messages.
+	for i := 0; i < 5; i++ {
+		_, err := l.Append([]*proto.Message{&proto.Message{
+			Value:       []byte(strconv.Itoa(i)),
+			Timestamp:   time.Now().UnixNano(),
+			LeaderEpoch: 1,
+		}})
+		require.NoError(t, err)
+	}
+
+	// Get current log end offset.
+	leo := l.NewestOffset()
+
+	// Register a waiter.
+	waiter := struct{}{}
+	ch := l.NotifyLEO(waiter, leo)
+
+	select {
+	case <-ch:
+		t.Fatalf("Unexpected channel close")
+	default:
+	}
+
+	// Add another message.
+	_, err := l.Append([]*proto.Message{&proto.Message{
+		Value:       []byte(strconv.Itoa(5)),
+		Timestamp:   time.Now().UnixNano(),
+		LeaderEpoch: 1,
+	}})
+	require.NoError(t, err)
+
+	select {
+	case <-ch:
+	default:
+		t.Fatalf("Expected channel to close")
+	}
+}
+
+// Ensure NotifyLEO returns the same channel if the waiter is already
+// registered in the log.
+func TestNotifyLEOIdempotent(t *testing.T) {
+	l, cleanup := setupWithOptions(t, Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 256,
+	})
+	defer l.Close()
+	defer cleanup()
+
+	// Get current log end offset.
+	leo := l.NewestOffset()
+
+	// Register a waiter.
+	waiter := struct{}{}
+	ch1 := l.NotifyLEO(waiter, leo)
+
+	// Register the same waiter again. This should return the same channel.
+	ch2 := l.NotifyLEO(waiter, leo)
+
+	require.Equal(t, ch1, ch2)
+}
+
 func setup(t require.TestingT) (*CommitLog, func()) {
 	opts := Options{
 		Path:            tempDir(t),
diff --git a/server/commitlog/segment_test.go b/server/commitlog/segment_test.go
index adf49b0..f5dd041 100644
--- a/server/commitlog/segment_test.go
+++ b/server/commitlog/segment_test.go
@@ -88,7 +88,7 @@ func TestSegmentSeal(t *testing.T) {
 	require.Equal(t, int64(10485760), stats.Size())
 
 	// Add a waiter.
-	ch := s.waitForData(&mockContextReader{}, 0)
+	ch := s.WaitForData(&mockContextReader{}, 0)
 
 	s.Seal()
 
diff --git a/server/partition_test.go b/server/partition_test.go
index 6c3d6ea..bb6aa6b 100644
--- a/server/partition_test.go
+++ b/server/partition_test.go
@@ -506,3 +506,68 @@ func TestPartitionAddToISRRecoverMin(t *testing.T) {
 	require.ElementsMatch(t, []string{"a", "b", "c"}, p.GetISR())
 	require.False(t, p.belowMinISR)
 }
+
+// Ensure replicationRequestLoop's idle follower sleep is preempted when a
+// partition notification is received.
+func TestPartitionReplicationRequestLoopPreempt(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Start NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Create NATS connection.
+	nc, err := nats.GetDefaultOptions().Connect()
+	require.NoError(t, err)
+	defer nc.Close()
+
+	// Start Liftbridge server.
+	config := getTestConfig("a", true, 5050)
+	// Set idle wait long enough that it shouldn't be reached.
+	config.Clustering.ReplicaMaxIdleWait = time.Hour
+	server := runServerWithConfig(t, config)
+	defer server.Stop()
+
+	p, err := server.newPartition(&proto.Partition{
+		Subject:  "foo",
+		Stream:   "foo",
+		Replicas: []string{"a", "b"},
+		Leader:   "b",
+		Isr:      []string{"a", "b"},
+	}, false)
+	require.NoError(t, err)
+	defer p.Close()
+
+	leaderEpoch := uint64(1)
+
+	// Set up mock leader.
+	requests := make(chan struct{}, 2)
+	_, err = nc.Subscribe(p.getReplicationRequestInbox(), func(msg *nats.Msg) {
+		// Send empty replication response.
+		resp := make([]byte, 16)
+		proto.Encoding.PutUint64(resp, leaderEpoch)
+		proto.Encoding.PutUint64(resp[8:], 0)
+		msg.Respond(resp)
+		requests <- struct{}{}
+	})
+	require.NoError(t, err)
+
+	stop := make(chan struct{})
+	defer close(stop)
+	go p.replicationRequestLoop("b", leaderEpoch, stop)
+
+	select {
+	case <-requests:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Expected replication request")
+	}
+
+	// Notify partition to preempt sleep.
+	p.Notify()
+
+	select {
+	case <-requests:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Expected replication request")
+	}
+}
diff --git a/server/replicator_test.go b/server/replicator_test.go
index 0cd8eb0..c5e8705 100644
--- a/server/replicator_test.go
+++ b/server/replicator_test.go
@@ -8,10 +8,12 @@ import (
 	lift "github.com/liftbridge-io/go-liftbridge"
 	"github.com/liftbridge-io/liftbridge-grpc/go"
 	natsdTest "github.com/nats-io/nats-server/v2/test"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
 
 	"github.com/liftbridge-io/liftbridge/server/commitlog"
+	internal "github.com/liftbridge-io/liftbridge/server/proto"
 )
 
 func waitForHW(t *testing.T, timeout time.Duration, name string, partitionID int32, hw int64, servers ...*Server) {
@@ -81,22 +83,25 @@ func TestStreamLeaderFailover(t *testing.T) {
 
 	// Configure first server.
 	s1Config := getTestConfig("a", true, 5050)
-	s1Config.Clustering.ReplicaMaxLeaderTimeout = 2 * time.Second
-	s1Config.Clustering.ReplicaFetchTimeout = time.Second
+	s1Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s1Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
+	s1Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
 
 	// Configure second server.
 	s2Config := getTestConfig("b", false, 5051)
-	s2Config.Clustering.ReplicaMaxLeaderTimeout = 2 * time.Second
-	s2Config.Clustering.ReplicaFetchTimeout = time.Second
+	s2Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s2Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
+	s2Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s2 := runServerWithConfig(t, s2Config)
 	defer s2.Stop()
 
 	// Configure second server.
 	s3Config := getTestConfig("c", false, 5052)
-	s3Config.Clustering.ReplicaMaxLeaderTimeout = 2 * time.Second
-	s3Config.Clustering.ReplicaFetchTimeout = time.Second
+	s3Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s3Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
+	s3Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s3 := runServerWithConfig(t, s3Config)
 	defer s3.Stop()
 
@@ -461,6 +466,7 @@ func TestTruncateFastLeaderElection(t *testing.T) {
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.Clustering.MinISR = 1
 	s1Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s1Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
 	s1Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
@@ -469,6 +475,7 @@ func TestTruncateFastLeaderElection(t *testing.T) {
 	s2Config := getTestConfig("b", false, 5051)
 	s2Config.Clustering.MinISR = 1
 	s2Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s2Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
 	s2Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s2 := runServerWithConfig(t, s2Config)
 	defer s2.Stop()
@@ -477,6 +484,7 @@ func TestTruncateFastLeaderElection(t *testing.T) {
 	s3Config := getTestConfig("c", false, 5052)
 	s3Config.Clustering.MinISR = 1
 	s3Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s3Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
 	s3Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s3 := runServerWithConfig(t, s3Config)
 	defer s3.Stop()
@@ -590,6 +598,7 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	s1Config := getTestConfig("a", true, 5050)
 	s1Config.Clustering.MinISR = 1
 	s1Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s1Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
 	s1Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
@@ -598,6 +607,7 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	s2Config := getTestConfig("b", false, 5051)
 	s2Config.Clustering.MinISR = 1
 	s2Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s2Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
 	s2Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s2 := runServerWithConfig(t, s2Config)
 	defer s2.Stop()
@@ -606,6 +616,7 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 	s3Config := getTestConfig("c", false, 5052)
 	s3Config.Clustering.MinISR = 1
 	s3Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
+	s3Config.Clustering.ReplicaMaxIdleWait = 500 * time.Millisecond
 	s3Config.Clustering.ReplicaFetchTimeout = 500 * time.Millisecond
 	s3 := runServerWithConfig(t, s3Config)
 	defer s3.Stop()
@@ -765,3 +776,80 @@ func TestTruncatePreventReplicaDivergence(t *testing.T) {
 		require.Equal(t, []byte("moon"), msg.Value())
 	}
 }
+
+// Ensure when a follower has caught up with the leader's log, the leader
+// notifies the follower when new messages are written to the partition.
+func TestReplicatorNotifyNewData(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Create NATS connection.
+	nc, err := nats.GetDefaultOptions().Connect()
+	require.NoError(t, err)
+	defer nc.Close()
+
+	// Configure first server.
+	s1Config := getTestConfig("a", true, 5050)
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	// Configure second server.
+	s2Config := getTestConfig("b", false, 5051)
+	s2 := runServerWithConfig(t, s2Config)
+	defer s2.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1, s2)
+
+	client, err := lift.Connect([]string{"localhost:5050", "localhost:5051"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	// Create stream.
+	name := "foo"
+	subject := "foo"
+	err = client.CreateStream(context.Background(), subject, name,
+		lift.ReplicationFactor(2))
+	require.NoError(t, err)
+
+	// Get partition leader.
+	leader := getPartitionLeader(t, 5*time.Second, name, 0, s1, s2)
+	var follower *Server
+	if leader == s1 {
+		follower = s2
+	} else {
+		follower = s1
+	}
+
+	// At this point, the follower is caught up with the leader since there
+	// aren't any messages. Set up a NATS subscription to intercept
+	// notifications.
+	var (
+		notifications = make(chan *internal.PartitionNotification)
+		inbox         = follower.getPartitionNotificationInbox(
+			follower.config.Clustering.ServerID)
+	)
+	_, err = nc.Subscribe(inbox, func(msg *nats.Msg) {
+		req := &internal.PartitionNotification{}
+		if err := req.Unmarshal(msg.Data); err != nil {
+			t.Fatalf("Invalid partition notification: %v", err)
+		}
+		notifications <- req
+	})
+	require.NoError(t, err)
+
+	// Publish a message. This will cause a notification to be sent to the
+	// follower.
+	_, err = client.Publish(context.Background(), subject, []byte("hello"))
+	require.NoError(t, err)
+
+	select {
+	case req := <-notifications:
+		require.Equal(t, name, req.Stream)
+		require.Equal(t, int32(0), req.Partition)
+	case <-time.After(5 * time.Second):
+		t.Fatal("Expected partition notification")
+	}
+}