Repository: liftbridge-io/liftbridge
Pull request #261 (2020-09-15 21:37:21)

Readonly streams/partitions
Builds on #237


Patch:
diff --git a/documentation/client_implementation.md b/documentation/client_implementation.md
index 8624046..f817dd4 100644
--- a/documentation/client_implementation.md
+++ b/documentation/client_implementation.md
@@ -28,22 +28,20 @@ to implementing a high-level client.
 
 ## Client Interface
 
-A high-level client has several operations:
-
-- [`CreateStream`](#createstream): creates a new stream attached to a NATS
-  subject (or group of related NATS subjects if partitioned)
-- [`DeleteStream`](#deletestream): deletes a stream and all of its partitions
-- [`PauseStream`](#pausestream): pauses some or all of a stream's partitions
-  until they are published to
-- [`Subscribe`](#subscribe): creates an ephemeral subscription for a given
-  stream that messages are received on
-- [`Publish`](#publish): publishes a new message to a Liftbridge stream
-- [`PublishAsync`](#publishasync): publishes a new message to a Liftbridge
-  stream asynchronously
-- [`PublishToSubject`](#publishtosubject): publishes a new message to a NATS
-  subject
-- [`FetchMetadata`](#fetchmetadata): retrieves metadata from the cluster
-- [`Close`](#close): closes any client connections to Liftbridge
+A high-level client has the following operations:
+
+| Operation | Description |
+|:----|:----|
+| [CreateStream](#createstream) | Creates a new stream attached to a NATS subject (or group of related NATS subjects if partitioned) |
+| [DeleteStream](#deletestream) | Deletes a stream and all of its partitions |
+| [PauseStream](#pausestream) | Pauses some or all of a stream's partitions until they are published to |
+| [SetStreamReadonly](#setstreamreadonly) | Sets some or all of a stream's partitions as readonly or readwrite |
+| [Subscribe](#subscribe) | Creates an ephemeral subscription for a given stream that messages are received on |
+| [Publish](#publish) | Publishes a new message to a Liftbridge stream |
+| [PublishAsync](#publishasync) | Publishes a new message to a Liftbridge stream asynchronously |
+| [PublishToSubject](#publishtosubject) | Publishes a new message to a NATS subject |
+| [FetchMetadata](#fetchmetadata) | Retrieves metadata from the cluster |
+| [Close](#close) | Closes any client connections to Liftbridge |
 
 Below is the interface definition of the Go Liftbridge client. We'll walk
 through each of these operations in more detail below.
@@ -73,6 +71,15 @@ type Client interface {
 	// in the stream is published to.
 	PauseStream(ctx context.Context, name string, opts ...PauseOption) error
 
+	// SetStreamReadonly sets the readonly flag on a stream and some or all of
+	// its partitions. Name is the stream identifier, globally unique. It
+	// returns an ErrNoSuchPartition if the given stream or partition does not
+	// exist. By default, this will set the readonly flag on all partitions.
+	// Subscribers to a readonly partition will see their subscription ended
+	// with a ErrReadonlyPartition error once all messages currently in the
+	// partition have been read.
+	SetStreamReadonly(ctx context.Context, name string, opts ...ReadonlyOption) error
+
 	// Subscribe creates an ephemeral subscription for the given stream. It
 	// begins receiving messages starting at the configured position and waits
 	// for new messages when it reaches the end of the stream. The default
@@ -244,6 +251,46 @@ configure stream pausing. Supported options are:
 
 [Implementation Guidance](#pausestream-implementation)
 
+### SetStreamReadonly
+
+```go
+// SetStreamReadonly sets the readonly flag on a stream and some or all of
+// its partitions. Name is the stream identifier, globally unique. It
+// returns an ErrNoSuchPartition if the given stream or partition does not
+// exist. By default, this will set the readonly flag on all partitions.
+// Subscribers to a readonly partition will see their subscription ended
+// with a ErrReadonlyPartition error once all messages currently in the
+// partition have been read.
+func (c *client) SetStreamReadonly(ctx context.Context, name string, options ...ReadonlyOption) error
+```
+
+`SetStreamReadonly` sets some or all of a stream's partitions as readonly or
+readwrite. Attempting to publish to a readonly partition will result in an
+`ErrReadonlyPartition` error/exception. Subscribers consuming a readonly
+partition will read up to the end of the partition and then receive an
+`ErrReadonlyPartition` error.
+
+In the Go client example above, `SetStreamReadonly` takes three arguments:
+
+| Argument | Type | Description | Required |
+|:----|:----|:----|:----|
+| context | context | A [context](https://golang.org/pkg/context/#Context) which is a Go idiom for passing things like a timeout, cancellation signal, and other values across API boundaries. For Liftbridge, this is primarily used for two things: request timeouts and cancellation. In other languages, this might be replaced by explicit arguments, optional named arguments, or other language-specific idioms. | language-dependent |
+| name | string | The name of the stream whose partitions to set the readonly flag for. | yes |
+| options | readonly options | Zero or more readonly options. These are used to pass in optional settings for setting the readonly flag. This is a common [Go pattern](https://dave.cheney.net/2014/10/17/functional-options-for-friendly-apis) for implementing extensible APIs. In other languages, this might be replaced with a builder pattern or optional named arguments. These `SetStreamReadonly` options are described below. | language-dependent |
+
+The readonly options are the equivalent of optional named arguments used to
+configure readonly requests. Supported options are:
+
+| Option | Type | Description | Default |
+|:----|:----|:----|:----|
+| ReadonlyPartitions | list of ints | Specifies the stream partitions to set the readonly flag for. If not set, all partitions will be changed. | |
+| Readonly | bool | Defines if the partitions should be set to readonly or readwrite. | false |
+
+`SetStreamReadonly` returns/throws an error if the operation fails,
+specifically `ErrNoSuchPartition` if the stream or partition(s) do not exist.
+
+[Implementation Guidance](#setstreamreadonly-implementation)
+
 ### Subscribe
 
 ```go
@@ -863,6 +910,8 @@ func asyncErrorFromProto(asyncError *proto.PublishAsyncError) error {
 	switch asyncError.Code {
 	case proto.PublishAsyncError_NOT_FOUND:
 		return ErrNoSuchPartition
+	case proto.PublishAsyncError_READONLY:
+		return ErrReadonlyPartition
 	default:
 		return errors.New(asyncError.Message)
 	}
@@ -886,6 +935,7 @@ idempotency reasons. This resilient RPC method can be used for RPCs such as:
 - `CreateStream`
 - `DeleteStream`
 - `PauseStream`
+- `SetStreamReadonly`
 - `PublishToSubject`
 - `FetchMetadata`
 
@@ -1116,6 +1166,46 @@ func (c *client) PauseStream(ctx context.Context, name string, options ...PauseO
 }
 ```
 
+### SetStreamReadonly Implementation
+
+The `SetStreamReadonly` implementation simply constructs a gRPC request and
+executes it using the [resilient RPC method](#rpcs) described above. If the
+`NotFound` gRPC error is returned, an `ErrNoSuchPartition` error/exception is
+thrown.  Otherwise, any other error/exception is thrown if the operation
+failed.
+
+```go
+// SetStreamReadonly sets the readonly flag on a stream and some or all of
+// its partitions. Name is the stream identifier, globally unique. It
+// returns an ErrNoSuchPartition if the given stream or partition does not
+// exist. By default, this will set the readonly flag on all partitions.
+// Subscribers to a readonly partition will see their subscription ended
+// with a ErrReadonlyPartition error once all messages currently in the
+// partition have been read.
+func (c *client) SetStreamReadonly(ctx context.Context, name string, options ...ReadonlyOption) error {
+	opts := &ReadonlyOptions{}
+	for _, opt := range options {
+		if err := opt(opts); err != nil {
+			return err
+		}
+	}
+
+	req := &proto.SetStreamReadonlyRequest{
+		Name:       name,
+		Partitions: opts.Partitions,
+		Readonly:   !opts.Readwrite,
+	}
+	err := c.doResilientRPC(func(client proto.APIClient) error {
+		_, err := client.SetStreamReadonly(ctx, req)
+		return err
+	})
+	if status.Code(err) == codes.NotFound {
+		return ErrNoSuchPartition
+	}
+	return err
+}
+```
+
 ### Subscribe Implementation
 
 `Subscribe` works by making a gRPC request which returns a stream that sends
diff --git a/go.mod b/go.mod
index ba1b534..916692c 100644
--- a/go.mod
+++ b/go.mod
@@ -14,8 +14,8 @@ require (
 	github.com/hashicorp/go-immutable-radix v1.2.0 // indirect
 	github.com/hashicorp/go-msgpack v1.1.5 // indirect
 	github.com/hashicorp/raft v1.1.2
-	github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20200925161903-f51717f56d5b
-	github.com/liftbridge-io/liftbridge-api v1.1.1-0.20200924210717-4ee8bde00417
+	github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20200925190539-82cd3725fbb6
+	github.com/liftbridge-io/liftbridge-api v1.1.1-0.20200925184350-33638578d994
 	github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3
 	github.com/liftbridge-io/raft-boltdb v0.0.0-20200414234651-aaf6e08d8f73
 	github.com/mattn/go-colorable v0.1.7 // indirect
@@ -39,8 +39,6 @@ require (
 	github.com/urfave/cli v1.22.4
 	go.etcd.io/bbolt v1.3.5 // indirect
 	golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de // indirect
-	golang.org/x/net v0.0.0-20200925080053-05aa5d4ee321 // indirect
-	google.golang.org/genproto v0.0.0-20200925023002-c2d885f95484 // indirect
 	google.golang.org/grpc v1.32.0
 	gopkg.in/ini.v1 v1.57.0 // indirect
 	launchpad.net/gocheck v0.0.0-20140225173054-000000000087 // indirect
diff --git a/go.sum b/go.sum
index 8ec4a93..91c283c 100644
--- a/go.sum
+++ b/go.sum
@@ -183,10 +183,10 @@ github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORN
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20200925161903-f51717f56d5b h1:gjUmUu60/se7/0Wxc6Q6dKIwvyCxNvfpYrw7OsH9DdE=
-github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20200925161903-f51717f56d5b/go.mod h1:+W+CfuDwHvoqJs58kfFlpGFSbwokRmFjXbiXp609yss=
-github.com/liftbridge-io/liftbridge-api v1.1.1-0.20200924210717-4ee8bde00417 h1:fqIq/1dG0mGUWdvNvScpbxyLoYIQsPnNgwOuRuAr9Xo=
-github.com/liftbridge-io/liftbridge-api v1.1.1-0.20200924210717-4ee8bde00417/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20200925190539-82cd3725fbb6 h1:/AlW3r0ak4Jo5w+bMSyHlLBYedzcEqzfbEGd5IPf05A=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20200925190539-82cd3725fbb6/go.mod h1:QbHIDs9udiRoYfW9ud0bRXAE3WiWZYTwv2IbSTZGpX8=
+github.com/liftbridge-io/liftbridge-api v1.1.1-0.20200925184350-33638578d994 h1:te6xY1NmRUHtKufqQ9lTBVMJQ5pjZDhk/MmeINuLQqQ=
+github.com/liftbridge-io/liftbridge-api v1.1.1-0.20200925184350-33638578d994/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
 github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3 h1:O4mg1NEmukgY8hxen3grrG5RY34LadMTzpbjf8kM2tA=
 github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3/go.mod h1:wmIIYVq+psahPlB1rvtTkGiltdihsKJbqwE1DkIPwj4=
 github.com/liftbridge-io/raft-boltdb v0.0.0-20200414234651-aaf6e08d8f73 h1:8r/ReB1ns87pVDwSnPj87HIbOu/5y0uDyGChx9mUGSQ=
@@ -195,7 +195,6 @@ github.com/magiconair/properties v1.8.1 h1:ZC2Vc7/ZFkGmsVC9KvOjumD+G5lXy2RtTKyzR
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
 github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
-github.com/mattn/go-colorable v0.1.6 h1:6Su7aK7lXmJ/U79bYtBjLNaha4Fs1Rg9plHpcH+vvnE=
 github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.7 h1:bQGKb3vps/j0E9GfJQ03JyhRuxsvdAanXlT9BTw3mdw=
 github.com/mattn/go-colorable v0.1.7/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
@@ -213,7 +212,6 @@ github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eI
 github.com/mitchellh/gox v0.4.0/go.mod h1:Sd9lOJ0+aimLBi73mGofS1ycjY8lL3uZM3JPS42BGNg=
 github.com/mitchellh/iochan v1.0.0/go.mod h1:JwYml1nuB7xOzsp52dPpHFffvOCDupsG0QubkSMEySY=
 github.com/mitchellh/mapstructure v0.0.0-20160808181253-ca63d7c062ee/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
-github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/mitchellh/mapstructure v1.3.2 h1:mRS76wmkOn3KkKAyXDu42V+6ebnXWIztFSYGN7GeoRg=
 github.com/mitchellh/mapstructure v1.3.2/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
@@ -225,20 +223,17 @@ github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRW
 github.com/natefinch/atomic v0.0.0-20200526193002-18c0533a5b09 h1:DXR0VtCesBD2ss3toN9OEeXszpQmW9dc3SvUbUfiBC0=
 github.com/natefinch/atomic v0.0.0-20200526193002-18c0533a5b09/go.mod h1:1rLVY/DWf3U6vSZgH16S7pymfrhK2lcUlXjgGglw/lY=
 github.com/nats-io/jwt v0.3.0/go.mod h1:fRYCDE99xlTsqUzISS1Bi75UBJ6ljOJQOAAu5VglpSg=
-github.com/nats-io/jwt v0.3.2 h1:+RB5hMpXUUA2dfxuhBTEkMOrYmM+gKIZYS1KjSostMI=
 github.com/nats-io/jwt v0.3.2/go.mod h1:/euKqTS1ZD+zzjYrY7pseZrTtWQSjujC7xjPc8wL6eU=
 github.com/nats-io/jwt v1.0.1 h1:71ivoESdfT2K/qDiw5YwX/3W9/dR7c+m83xiGOj/EZ4=
 github.com/nats-io/jwt v1.0.1/go.mod h1:n3cvmLfBfnpV4JJRN7lRYCyZnw48ksGsbThGXEk4w9M=
 github.com/nats-io/nats-server/v2 v2.1.4 h1:BILRnsJ2Yb/fefiFbBWADpViGF69uh4sxe8poVDQ06g=
 github.com/nats-io/nats-server/v2 v2.1.4/go.mod h1:Jw1Z28soD/QasIA2uWjXyM9El1jly3YwyFOuR8tH1rg=
 github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzEE/Zbp4w=
-github.com/nats-io/nats.go v1.9.2 h1:oDeERm3NcZVrPpdR/JpGdWHMv3oJ8yY30YwxKq+DU2s=
 github.com/nats-io/nats.go v1.9.2/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nats.go v1.10.0 h1:L8qnKaofSfNFbXg0C5F71LdjPRnmQwSsA4ukmkt1TvY=
 github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
-github.com/nats-io/nkeys v0.1.4 h1:aEsHIssIk6ETN5m2/MD8Y4B2X7FfXrBAUdkyRvbVYzA=
 github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.2.0 h1:WXKF7diOaPU9cJdLD7nuzwasQy9vT1tBqzXZZf3AMJM=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
@@ -250,7 +245,6 @@ github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY=
 github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
-github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pelletier/go-toml v1.8.0 h1:Keo9qb7iRJs2voHvunFtuuYFsbWeOBh8/P9v/kVMFtw=
 github.com/pelletier/go-toml v1.8.0/go.mod h1:D6yutnOGMveHEPV7VQOuvI/gXY61bv+9bAOTRnLElKs=
@@ -300,19 +294,15 @@ github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIK
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
-github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.3.1 h1:GPTpEAuNr98px18yNQ66JllNil98wfRZ/5Ukny8FeQA=
 github.com/spf13/afero v1.3.1/go.mod h1:5KUK8ByomD5Ti5Artl0RtHeI5pTF7MIDuXL3yY520V4=
-github.com/spf13/cast v1.3.0 h1:oget//CVOEoFewqQxwr0Ej5yjygnqGkvggSE/gB35Q8=
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
-github.com/spf13/jwalterweatherman v1.0.0 h1:XHEdyB+EcvlqZamSM4ZOMGlc93t6AcsBEu9Gc1vn7yk=
 github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb68N+wFjFa4jdeBTo=
 github.com/spf13/jwalterweatherman v1.1.0 h1:ue6voC5bR5F8YxI5S67j9i582FU4Qvo2bmqnqMYADFk=
 github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
-github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
@@ -332,7 +322,6 @@ github.com/urfave/cli v1.22.4 h1:u7tSpNPPswAFymm8IehJhy4uJMlUuU/GmqSkvJ1InXA=
 github.com/urfave/cli v1.22.4/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
-go.etcd.io/bbolt v1.3.4 h1:hi1bXHMVrlQh6WwxAy+qZCV/SYIlqo+Ushwdpa4tAKg=
 go.etcd.io/bbolt v1.3.4/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
@@ -350,7 +339,6 @@ golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de h1:ikNHVSjEfnvz6sxdSPCaPt572qowuyMDMJLLm3Db3ig=
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -387,8 +375,6 @@ golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200923182212-328152dc79b1 h1:Iu68XRPd67wN4aRGGWwwq6bZo/25jR6uu52l/j2KkUE=
-golang.org/x/net v0.0.0-20200923182212-328152dc79b1/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200925080053-05aa5d4ee321 h1:lleNcKRbcaC8MqgLwghIkzZ2JBQAb7QQ9MiwRt1BisA=
 golang.org/x/net v0.0.0-20200925080053-05aa5d4ee321/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -431,7 +417,6 @@ golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d h1:L/IKR6COd7ubZrs2oTnTi73Ih
 golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -478,8 +463,6 @@ google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98
 google.golang.org/genproto v0.0.0-20190911173649-1774047e7e51/go.mod h1:IbNlFCBrqXvoKpeg0TB2l7cyZUmoaFKYIwrEpbDKLA8=
 google.golang.org/genproto v0.0.0-20191108220845-16a3f7862a1a/go.mod h1:n3cpQtvxv34hfy77yVDNjmbRyujviMdxYliBSkLhpCc=
 google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013/go.mod h1:NbSheEEYHJ7i3ixzK3sjbqSGDJWnxyFXZblF3eUsNvo=
-google.golang.org/genproto v0.0.0-20200924141100-a14c0a98937d h1:gQhS6YjPEa6qyVIEcEC2oYp5T6vRgv7kJhSqU9Cqhiw=
-google.golang.org/genproto v0.0.0-20200924141100-a14c0a98937d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20200925023002-c2d885f95484 h1:Rr9EZdYRq2WLckzJQVtN3ISKoP7dvgwi7jbglILNZ34=
 google.golang.org/genproto v0.0.0-20200925023002-c2d885f95484/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
@@ -496,10 +479,8 @@ google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQ
 google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
 google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
 google.golang.org/protobuf v1.22.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
 google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
 google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
-google.golang.org/protobuf v1.24.0 h1:UhZDfRO8JRQru4/+LlLE0BRKGF8L+PICnvYZmx/fEGA=
 google.golang.org/protobuf v1.24.0/go.mod h1:r/3tXBNzIEhYS9I1OUVjXDlt8tc493IdKGjtUeSXeh4=
 google.golang.org/protobuf v1.25.0 h1:Ejskq+SyPohKW+1uil0JJMtmHCgJPJ/qWTxr8qp+R4c=
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
@@ -509,7 +490,6 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
-gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno=
 gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/ini.v1 v1.57.0 h1:9unxIsFcTt4I55uWluz+UmL95q4kdJ0buvQ1ZIqVQww=
 gopkg.in/ini.v1 v1.57.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
@@ -519,7 +499,6 @@ gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/server/activity.go b/server/activity.go
index 852fd3e..c0a7c90 100644
--- a/server/activity.go
+++ b/server/activity.go
@@ -203,6 +203,15 @@ func (a *activityManager) handleRaftLog(l *raft.Log) error {
 				Partitions: log.ResumeStreamOp.Partitions,
 			},
 		}
+	case proto.Op_SET_STREAM_READONLY:
+		event = &client.ActivityStreamEvent{
+			Op: client.ActivityStreamOp_SET_STREAM_READONLY,
+			SetStreamReadonlyOp: &client.SetStreamReadonlyOp{
+				Stream:     log.SetStreamReadonlyOp.Stream,
+				Partitions: log.SetStreamReadonlyOp.Partitions,
+				Readonly:   log.SetStreamReadonlyOp.Readonly,
+			},
+		}
 	default:
 		return nil
 	}
diff --git a/server/api.go b/server/api.go
index 2560f46..b84a5e1 100644
--- a/server/api.go
+++ b/server/api.go
@@ -129,8 +129,31 @@ func (a *apiServer) PauseStream(ctx context.Context, req *client.PauseStreamRequ
 // readonly status set.
 func (a *apiServer) SetStreamReadonly(ctx context.Context, req *client.SetStreamReadonlyRequest) (
 	*client.SetStreamReadonlyResponse, error) {
-	// TODO
-	return nil, status.Error(codes.Unimplemented, "not yet implemented")
+
+	resp := &client.SetStreamReadonlyResponse{}
+	a.logger.Debugf("api: SetStreamReadonly [name=%s, partitions=%v, readonly=%v]",
+		req.Name, req.Partitions, req.Readonly)
+
+	if len(req.Partitions) == 0 {
+		stream := a.metadata.GetStream(req.Name)
+		if stream == nil {
+			return nil, status.Error(codes.NotFound, "stream not found")
+		}
+		for _, partition := range stream.GetPartitions() {
+			req.Partitions = append(req.Partitions, partition.Id)
+		}
+	}
+
+	if e := a.metadata.SetStreamReadonly(ctx, &proto.SetStreamReadonlyOp{
+		Stream:     req.Name,
+		Partitions: req.Partitions,
+		Readonly:   req.Readonly,
+	}); e != nil {
+		a.logger.Errorf("api: Failed to set stream readonly flag %v: %v", req.Name, e.Err())
+		return nil, e.Err()
+	}
+
+	return resp, nil
 }
 
 // Subscribe creates an ephemeral subscription for the given stream partition.
@@ -209,7 +232,8 @@ func (a *apiServer) FetchMetadata(ctx context.Context, req *client.FetchMetadata
 // Publish a new message to a stream. If the AckPolicy is not NONE and a
 // deadline is provided, this will synchronously block until the ack is
 // received. If the ack is not received in time, a DeadlineExceeded status code
-// is returned.
+// is returned. A FailedPrecondition status code is returned if the partition is
+// readonly.
 func (a *apiServer) Publish(ctx context.Context, req *client.PublishRequest) (
 	*client.PublishResponse, error) {
 
@@ -222,6 +246,10 @@ func (a *apiServer) Publish(ctx context.Context, req *client.PublishRequest) (
 		return nil, convertPublishAsyncError(e)
 	}
 
+	if e := a.ensureStreamNotReadonly(req.Stream, req.Partition); e != nil {
+		return nil, convertPublishAsyncError(e)
+	}
+
 	if err := a.resumeStream(ctx, req.Stream, req.Partition); err != nil {
 		a.logger.Errorf("api: Failed to resume stream: %v", err)
 		return nil, err
@@ -317,6 +345,50 @@ func (a *apiServer) PublishToSubject(ctx context.Context, req *client.PublishToS
 	return resp, nil
 }
 
+// SetCursor stores a cursor position for a particular stream partition which
+// is uniquely identified by an opaque string.
+//
+// NOTE: This is a beta endpoint and is subject to change. It is not included
+// as part of Liftbridge's semantic versioning scheme.
+func (a *apiServer) SetCursor(ctx context.Context, req *client.SetCursorRequest) (
+	*client.SetCursorResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "not implemented")
+}
+
+// FetchCursor retrieves a partition cursor position.
+//
+// NOTE: This is a beta endpoint and is subject to change. It is not included
+// as part of Liftbridge's semantic versioning scheme.
+func (a *apiServer) FetchCursor(ctx context.Context, req *client.FetchCursorRequest) (
+	*client.FetchCursorResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "not implemented")
+}
+
+func (a *apiServer) ensureStreamNotReadonly(name string, partitionID int32) *client.PublishAsyncError {
+	stream := a.metadata.GetStream(name)
+	if stream == nil {
+		return &client.PublishAsyncError{
+			Code:    client.PublishAsyncError_NOT_FOUND,
+			Message: fmt.Sprintf("no such stream: %s", name),
+		}
+	}
+	partition := stream.GetPartition(partitionID)
+	if partition == nil {
+		return &client.PublishAsyncError{
+			Code:    client.PublishAsyncError_NOT_FOUND,
+			Message: fmt.Sprintf("no such partition: %d", partitionID),
+		}
+	}
+	if partition.IsReadonly() {
+		return &client.PublishAsyncError{
+			Code:    client.PublishAsyncError_READONLY,
+			Message: fmt.Sprintf("readonly partition: %d", partitionID),
+		}
+	}
+
+	return nil
+}
+
 func (a *apiServer) resumeStream(ctx context.Context, streamName string, partitionID int32) error {
 	stream := a.metadata.GetStream(streamName)
 	if stream == nil {
@@ -393,6 +465,12 @@ func (a *apiServer) publishAsyncLoop(stream client.API_PublishAsyncServer, ackIn
 			return err
 		}
 
+		if e := a.ensureStreamNotReadonly(req.Stream, req.Partition); e != nil {
+			a.logger.Errorf("api: Failed to publish async message: %v", e.Message)
+			a.sendPublishAsyncError(stream, req.CorrelationId, e)
+			continue
+		}
+
 		req.AckInbox = ackInbox
 
 		a.logger.Debugf("api: PublishAsync [stream=%s, partition=%d]", req.Stream, req.Partition)
@@ -540,6 +618,8 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 		return nil, nil, st
 	}
 
+	endOffset := partition.log.NewestOffset()
+
 	var (
 		ch          = make(chan *client.Message)
 		errCh       = make(chan *status.Status)
@@ -567,6 +647,9 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 						code = codes.FailedPrecondition
 					}
 					s = status.New(code, err.Error())
+				} else if err == commitlog.ErrCommitLogReadonly {
+					// Partition was set to readonly while subscribed.
+					s = status.New(codes.ResourceExhausted, fmt.Sprintf("End of readonly partition"))
 				} else {
 					s = status.Convert(err)
 				}
@@ -596,6 +679,15 @@ func (a *apiServer) subscribe(ctx context.Context, partition *partition,
 			case <-cancel:
 				return
 			}
+			if partition.IsReadonly() && offset == endOffset {
+				s := status.New(codes.ResourceExhausted, fmt.Sprintf("End of readonly partition"))
+
+				select {
+				case errCh <- s:
+				case <-cancel:
+				}
+				return
+			}
 		}
 	})
 
@@ -673,8 +765,10 @@ func convertPublishAsyncError(err *client.PublishAsyncError) error {
 		code = codes.NotFound
 	case client.PublishAsyncError_BAD_REQUEST:
 		code = codes.InvalidArgument
+	case client.PublishAsyncError_READONLY:
+		code = codes.FailedPrecondition
 	case client.PublishAsyncError_UNKNOWN:
-		code = codes.Unknown
+		fallthrough
 	default:
 		code = codes.Unknown
 	}
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index e558dd1..0bd76d8 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -34,7 +34,7 @@ const (
 // commitLog implements the CommitLog interface, which is a durable write-ahead
 // log.
 type commitLog struct {
-	Options
+	readonly         int32 // Atomic flag
 	deleteCleaner    *deleteCleaner
 	compactCleaner   *compactCleaner
 	name             string
@@ -43,9 +43,10 @@ type commitLog struct {
 	closed           chan struct{}
 	segments         []*segment
 	vActiveSegment   *segment
-	hwWaiters        map[contextReader]chan struct{}
+	hwWaiters        map[contextReader]chan bool
 	leaderEpochCache *leaderEpochCache
 	deleted          bool
+	Options
 }
 
 // Options contains settings for configuring a commitLog.
@@ -115,7 +116,7 @@ func New(opts Options) (CommitLog, error) {
 		compactCleaner:   compactCleaner,
 		hw:               -1,
 		closed:           make(chan struct{}),
-		hwWaiters:        make(map[contextReader]chan struct{}),
+		hwWaiters:        make(map[contextReader]chan bool),
 		leaderEpochCache: epochCache,
 	}
 
@@ -211,8 +212,12 @@ func (l *commitLog) open() error {
 }
 
 // Append writes the given batch of messages to the log and returns their
-// corresponding offsets in the log.
+// corresponding offsets in the log. This will return ErrCommitLogReadonly if
+// the log is in readonly mode.
 func (l *commitLog) Append(msgs []*Message) ([]int64, error) {
+	if l.IsReadonly() {
+		return nil, ErrCommitLogReadonly
+	}
 	if _, err := l.checkAndPerformSplit(); err != nil {
 		return nil, err
 	}
@@ -229,7 +234,9 @@ func (l *commitLog) Append(msgs []*Message) ([]int64, error) {
 }
 
 // AppendMessageSet writes the given message set data to the log and returns
-// the corresponding offsets in the log.
+// the corresponding offsets in the log. This can be called even if the log is
+// in readonly mode to allow for reconciliation, e.g. when replicating from
+// another log.
 func (l *commitLog) AppendMessageSet(ms []byte) ([]int64, error) {
 	if _, err := l.checkAndPerformSplit(); err != nil {
 		return nil, err
@@ -327,7 +334,7 @@ func (l *commitLog) SetHighWatermark(hw int64) {
 	l.mu.Lock()
 	if hw > l.hw {
 		l.hw = hw
-		l.notifyHWWaiters()
+		l.notifyHWChange()
 	}
 	l.mu.Unlock()
 	// TODO: should we flush the HW to disk here?
@@ -339,24 +346,47 @@ func (l *commitLog) SetHighWatermark(hw int64) {
 func (l *commitLog) OverrideHighWatermark(hw int64) {
 	l.mu.Lock()
 	l.hw = hw
-	l.notifyHWWaiters()
+	l.notifyHWChange()
 	l.mu.Unlock()
 }
 
-func (l *commitLog) notifyHWWaiters() {
+// notifyHWChange signals all HW waiters to wake up because the HW has changed.
+// This must be called within the log mutex.
+func (l *commitLog) notifyHWChange() {
 	for r, ch := range l.hwWaiters {
-		close(ch)
+		ch <- false
 		delete(l.hwWaiters, r)
 	}
 }
 
-func (l *commitLog) waitForHW(r contextReader, hw int64) <-chan struct{} {
-	wait := make(chan struct{})
+// notifyReadonly signals all HW waiters to wake up if the HW is caught up to
+// the LEO because the log has become readonly. This must be called within the
+// log mutex.
+func (l *commitLog) notifyReadonly() {
+	if l.hw < l.NewestOffset() {
+		return
+	}
+	// HW is caught up to LEO so notify HW waiters.
+	for r, ch := range l.hwWaiters {
+		ch <- true
+		delete(l.hwWaiters, r)
+	}
+}
+
+// waitForHW registers an HW waiter and returns a channel which will receive a
+// bool either when the HW changes (false) or the log has become readonly
+// (true).
+func (l *commitLog) waitForHW(r contextReader, hw int64) <-chan bool {
+	wait := make(chan bool, 1)
 	l.mu.Lock()
-	// Check if HW has changed.
 	if l.hw != hw {
-		close(wait)
+		// HW has changed since reader last checked so they can unblock now.
+		wait <- false
+	} else if l.hw == l.NewestOffset() && l.IsReadonly() {
+		// Log is readonly and HW is caught up to LEO so return an error to reader.
+		wait <- true
 	} else {
+		// Reader needs to wait for HW to advance.
 		l.hwWaiters[r] = wait
 	}
 	l.mu.Unlock()
@@ -545,6 +575,32 @@ func (l *commitLog) NotifyLEO(waiter interface{}, leo int64) <-chan struct{} {
 	return l.activeSegment().WaitForLEO(waiter, leo)
 }
 
+// SetReadonly marks the log as readonly. When in readonly mode, new messages
+// cannot be added to the log with Append and committed readers will read up to
+// the log end offset (LEO), if the HW allows so, and then will receive an
+// ErrCommitLogReadonly error. This will unblock committed readers waiting for
+// data if they are at the LEO. Readers will continue to block if the HW is
+// less than the LEO. This does not affect uncommitted readers. Messages can
+// still be written to the log with AppendMessageSet for reconciliation
+// purposes, e.g. when replicating from another log.
+func (l *commitLog) SetReadonly(readonly bool) {
+	value := int32(0)
+	if readonly {
+		value = 1
+	}
+	atomic.StoreInt32(&l.readonly, value)
+	if readonly {
+		l.mu.Lock()
+		l.notifyReadonly()
+		l.mu.Unlock()
+	}
+}
+
+// IsReadonly indicates if the log is in readonly mode.
+func (l *commitLog) IsReadonly() bool {
+	return atomic.LoadInt32(&l.readonly) == 1
+}
+
 // checkAndPerformSplit determines if a new log segment should be rolled out
 // either because the active segment is full or MaxSegmentAge has passed since
 // the first message was written to it. It then performs the split if eligible,
diff --git a/server/commitlog/interface.go b/server/commitlog/interface.go
index 9058a24..4bfe4b7 100644
--- a/server/commitlog/interface.go
+++ b/server/commitlog/interface.go
@@ -50,11 +50,14 @@ type CommitLog interface {
 	LastLeaderEpoch() uint64
 
 	// Append writes the given batch of messages to the log and returns their
-	// corresponding offsets in the log.
+	// corresponding offsets in the log. This will return ErrCommitLogReadonly
+	// if the log is in readonly mode.
 	Append(msg []*Message) ([]int64, error)
 
 	// AppendMessageSet writes the given message set data to the log and
-	// returns the corresponding offsets in the log.
+	// returns the corresponding offsets in the log. This can be called even if
+	// the log is in readonly mode to allow for reconciliation, e.g. when
+	// replicating from another log.
 	AppendMessageSet(ms []byte) ([]int64, error)
 
 	// Clean applies retention and compaction rules against the log, if
@@ -68,6 +71,20 @@ type CommitLog interface {
 	// for data.
 	NotifyLEO(waiter interface{}, leo int64) <-chan struct{}
 
+	// SetReadonly marks the log as readonly. When in readonly mode, new
+	// messages cannot be added to the log with Append and committed readers
+	// will read up to the log end offset (LEO), if the HW allows so, and then
+	// will receive an ErrCommitLogReadonly error. This will unblock committed
+	// readers waiting for data if they are at the LEO. Readers will continue
+	// to block if the HW is less than the LEO. This does not affect
+	// uncommitted readers. Messages can still be written to the log with
+	// AppendMessageSet for reconciliation purposes, e.g. when replicating from
+	// another log.
+	SetReadonly(readonly bool)
+
+	// IsReadonly indicates if the log is in readonly mode.
+	IsReadonly() bool
+
 	// Close closes each log segment file and stops the background goroutine
 	// checkpointing the high watermark to disk.
 	Close() error
diff --git a/server/commitlog/reader.go b/server/commitlog/reader.go
index 0009f20..19ead77 100644
--- a/server/commitlog/reader.go
+++ b/server/commitlog/reader.go
@@ -9,6 +9,10 @@ import (
 	pkgErrors "github.com/pkg/errors"
 )
 
+// ErrCommitLogReadonly is returned when the end of a readonly CommitLog has
+// been reached.
+var ErrCommitLogReadonly = errors.New("end of readonly log")
+
 type contextReader interface {
 	Read(context.Context, []byte) (int, error)
 }
@@ -63,6 +67,9 @@ RETRY:
 		} else if r.log.IsClosed() {
 			// The log was closed while we were trying to read.
 			return nil, 0, 0, 0, ErrCommitLogClosed
+		} else if pkgErrors.Cause(err) == ErrCommitLogReadonly && r.log.IsReadonly() {
+			// The log was set to readonly while we were trying to read.
+			return nil, 0, 0, 0, ErrCommitLogReadonly
 		} else if pkgErrors.Cause(err) == ErrSegmentReplaced {
 			// ErrSegmentReplaced indicates we attempted to read from a log
 			// segment that was replaced due to compaction, so reinitialize the
@@ -220,8 +227,8 @@ func (r *committedReader) Read(ctx context.Context, p []byte) (n int, err error)
 		hw := r.cl.HighWatermark()
 		for hw == r.hw {
 			// The HW has not changed, so wait for it to update.
-			if !r.waitForHW(ctx, hw) {
-				err = io.EOF
+			err = r.waitForHW(ctx, hw)
+			if err != nil {
 				return
 			}
 			// Sync the HW.
@@ -290,8 +297,8 @@ LOOP:
 		hw := r.cl.HighWatermark()
 		for hw == r.hw {
 			// The HW has not changed, so wait for it to update.
-			if !r.waitForHW(ctx, hw) {
-				err = io.EOF
+			err = r.waitForHW(ctx, hw)
+			if err != nil {
 				break LOOP
 			}
 			// Sync the HW.
@@ -310,17 +317,20 @@ LOOP:
 	return n, err
 }
 
-func (r *committedReader) waitForHW(ctx context.Context, hw int64) bool {
+func (r *committedReader) waitForHW(ctx context.Context, hw int64) error {
 	wait := r.cl.waitForHW(r, hw)
 	select {
 	case <-r.cl.closed:
 		r.cl.removeHWWaiter(r)
-		return false
+		return io.EOF
 	case <-ctx.Done():
 		r.cl.removeHWWaiter(r)
-		return false
-	case <-wait:
-		return true
+		return io.EOF
+	case readonly := <-wait:
+		if readonly {
+			return ErrCommitLogReadonly
+		}
+		return nil
 	}
 }
 
diff --git a/server/fsm.go b/server/fsm.go
index eb3ef59..b5da54f 100644
--- a/server/fsm.go
+++ b/server/fsm.go
@@ -174,6 +174,15 @@ func (s *Server) apply(log *proto.RaftLog, index uint64, recovered bool) (interf
 		if err := s.applyPauseStream(stream, partitions, resumeAll); err != nil {
 			return nil, err
 		}
+	case proto.Op_SET_STREAM_READONLY:
+		var (
+			stream     = log.SetStreamReadonlyOp.Stream
+			partitions = log.SetStreamReadonlyOp.Partitions
+			readonly   = log.SetStreamReadonlyOp.Readonly
+		)
+		if err := s.applySetStreamReadonly(stream, partitions, readonly); err != nil {
+			return nil, err
+		}
 	case proto.Op_RESUME_STREAM:
 		var (
 			stream     = log.ResumeStreamOp.Stream
@@ -473,3 +482,20 @@ func (s *Server) applyResumeStream(streamName string, partitionIDs []int32, reco
 	}
 	return nil
 }
+
+// applySetStreamReadonly changes the stream partitions readonly flag in the
+// metadata store.
+func (s *Server) applySetStreamReadonly(streamName string, partitions []int32, readonly bool) error {
+	stream := s.metadata.GetStream(streamName)
+	if stream == nil {
+		return ErrStreamNotFound
+	}
+
+	err := stream.SetReadonly(partitions, readonly)
+	if err != nil {
+		return errors.Wrap(err, "failed to set stream as readonly")
+	}
+
+	s.logger.Debugf("fsm: Set stream %s readonly flag as %v", streamName, readonly)
+	return nil
+}
diff --git a/server/metadata.go b/server/metadata.go
index de14176..35da21f 100644
--- a/server/metadata.go
+++ b/server/metadata.go
@@ -631,6 +631,45 @@ func (m *metadataAPI) ReportLeader(ctx context.Context, req *proto.ReportLeaderO
 	return reported.addWitness(req.Replica)
 }
 
+// SetStreamReadonly sets a stream's readonly flag if this server is the
+// metadata leader. If it is not, it will forward the request to the leader and
+// return the response. This operation is replicated by Raft. If successful,
+// this will return once the readonly flag has been set.
+func (m *metadataAPI) SetStreamReadonly(ctx context.Context, req *proto.SetStreamReadonlyOp) *status.Status {
+	// Forward the request if we're not the leader.
+	if !m.IsLeader() {
+		isLeader, st := m.propagateSetStreamReadonly(ctx, req)
+		if st != nil {
+			return st
+		}
+		// If we have since become leader, continue on with the request.
+		if !isLeader {
+			return nil
+		}
+	}
+
+	// Replicate stream the readonly flag through Raft.
+	op := &proto.RaftLog{
+		Op:                  proto.Op_SET_STREAM_READONLY,
+		SetStreamReadonlyOp: req,
+	}
+
+	// Wait on result of setting the readonly flag.
+	future, err := m.getRaft().applyOperation(ctx, op, m.checkSetStreamReadonlyPreconditions)
+	if err != nil {
+		code := codes.FailedPrecondition
+		if err == ErrStreamNotFound || err == ErrPartitionNotFound {
+			code = codes.NotFound
+		}
+		return status.Newf(code, err.Error())
+	}
+	if err := future.Error(); err != nil {
+		return status.Newf(codes.Internal, "Failed to set stream readonly flag: %v", err.Error())
+	}
+
+	return nil
+}
+
 // AddStream adds the given stream and its partitions to the metadata store. It
 // returns an error if a stream with the same name or any partitions with the
 // same ID for the stream already exist. If the stream is recovered, this will
@@ -993,6 +1032,18 @@ func (m *metadataAPI) propagateReportLeader(ctx context.Context, req *proto.Repo
 	return m.propagateRequest(ctx, propagate)
 }
 
+// propagateSetStreamReadonly forwards a SetStreamReadonly request to the
+// metadata leader. The bool indicates if this server has since become leader
+// and the request should be performed locally. A Status is returned if the
+// propagated request failed.
+func (m *metadataAPI) propagateSetStreamReadonly(ctx context.Context, req *proto.SetStreamReadonlyOp) (bool, *status.Status) {
+	propagate := &proto.PropagatedRequest{
+		Op:                  proto.Op_SET_STREAM_READONLY,
+		SetStreamReadonlyOp: req,
+	}
+	return m.propagateRequest(ctx, propagate)
+}
+
 // propagateRequest forwards a metadata request to the metadata leader. The
 // bool indicates if this server has since become leader and the request should
 // be performed locally. A Status is returned if the propagated request failed.
@@ -1156,6 +1207,23 @@ func (m *metadataAPI) checkPauseStreamPreconditions(op *proto.RaftLog) error {
 	return nil
 }
 
+// checkSetStreamReadonlyPreconditions checks if the stream and partitions being
+// set readonly exist. If the stream doesn't exist, it returns
+// ErrStreamNotFound. If one or more specified partitions don't exist, it
+// returns ErrPartitionNotFound. Otherwise, it returns nil.
+func (m *metadataAPI) checkSetStreamReadonlyPreconditions(op *proto.RaftLog) error {
+	stream := m.GetStream(op.SetStreamReadonlyOp.Stream)
+	if stream == nil {
+		return ErrStreamNotFound
+	}
+	for _, partitionID := range op.SetStreamReadonlyOp.Partitions {
+		if partition := stream.GetPartition(partitionID); partition == nil {
+			return ErrPartitionNotFound
+		}
+	}
+	return nil
+}
+
 // checkResumeStreamPreconditions checks if the stream and partitions to be
 // resumed exist. If the stream does not exist, it returns ErrStreamNotFound.
 // If any partitions do not exist, it returns ErrPartitionNotFound. Otherwise,
diff --git a/server/partition.go b/server/partition.go
index dfac253..4b4b0ac 100644
--- a/server/partition.go
+++ b/server/partition.go
@@ -222,6 +222,18 @@ func (p *partition) IsPaused() bool {
 	return p.paused
 }
 
+// SetReadonly enables or disables readonly for the partition. When enabled,
+// new messages cannot be written to the log and consumers will not block once
+// they reach the end of the log. This does not affect replication.
+func (p *partition) SetReadonly(readonly bool) {
+	p.log.SetReadonly(readonly)
+}
+
+// IsReadonly indicates if the partition is currently readonly.
+func (p *partition) IsReadonly() bool {
+	return p.log.IsReadonly()
+}
+
 // Delete stops the partition if it is running, closes, and deletes the commit
 // log.
 func (p *partition) Delete() error {
diff --git a/server/protocol/internal.pb.go b/server/protocol/internal.pb.go
index e6f04d8..0b1c7ef 100644
--- a/server/protocol/internal.pb.go
+++ b/server/protocol/internal.pb.go
@@ -25,15 +25,16 @@ const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
 type Op int32
 
 const (
-	Op_CREATE_STREAM    Op = 0
-	Op_SHRINK_ISR       Op = 1
-	Op_REPORT_LEADER    Op = 2
-	Op_CHANGE_LEADER    Op = 3
-	Op_EXPAND_ISR       Op = 4
-	Op_DELETE_STREAM    Op = 5
-	Op_PAUSE_STREAM     Op = 6
-	Op_RESUME_STREAM    Op = 7
-	Op_PUBLISH_ACTIVITY Op = 8
+	Op_CREATE_STREAM       Op = 0
+	Op_SHRINK_ISR          Op = 1
+	Op_REPORT_LEADER       Op = 2
+	Op_CHANGE_LEADER       Op = 3
+	Op_EXPAND_ISR          Op = 4
+	Op_DELETE_STREAM       Op = 5
+	Op_PAUSE_STREAM        Op = 6
+	Op_RESUME_STREAM       Op = 7
+	Op_PUBLISH_ACTIVITY    Op = 8
+	Op_SET_STREAM_READONLY Op = 9
 )
 
 var Op_name = map[int32]string{
@@ -46,18 +47,20 @@ var Op_name = map[int32]string{
 	6: "PAUSE_STREAM",
 	7: "RESUME_STREAM",
 	8: "PUBLISH_ACTIVITY",
+	9: "SET_STREAM_READONLY",
 }
 
 var Op_value = map[string]int32{
-	"CREATE_STREAM":    0,
-	"SHRINK_ISR":       1,
-	"REPORT_LEADER":    2,
-	"CHANGE_LEADER":    3,
-	"EXPAND_ISR":       4,
-	"DELETE_STREAM":    5,
-	"PAUSE_STREAM":     6,
-	"RESUME_STREAM":    7,
-	"PUBLISH_ACTIVITY": 8,
+	"CREATE_STREAM":       0,
+	"SHRINK_ISR":          1,
+	"REPORT_LEADER":       2,
+	"CHANGE_LEADER":       3,
+	"EXPAND_ISR":          4,
+	"DELETE_STREAM":       5,
+	"PAUSE_STREAM":        6,
+	"RESUME_STREAM":       7,
+	"PUBLISH_ACTIVITY":    8,
+	"SET_STREAM_READONLY": 9,
 }
 
 func (x Op) String() string {
@@ -116,18 +119,19 @@ func (m *ServerState) GetServerID() string {
 }
 
 type RaftLog struct {
-	Op                   Op                 `protobuf:"varint,1,opt,name=op,proto3,enum=protocol.Op" json:"op,omitempty"`
-	CreateStreamOp       *CreateStreamOp    `protobuf:"bytes,2,opt,name=createStreamOp,proto3" json:"createStreamOp,omitempty"`
-	ShrinkISROp          *ShrinkISROp       `protobuf:"bytes,3,opt,name=shrinkISROp,proto3" json:"shrinkISROp,omitempty"`
-	ChangeLeaderOp       *ChangeLeaderOp    `protobuf:"bytes,4,opt,name=changeLeaderOp,proto3" json:"changeLeaderOp,omitempty"`
-	ExpandISROp          *ExpandISROp       `protobuf:"bytes,5,opt,name=expandISROp,proto3" json:"expandISROp,omitempty"`
-	DeleteStreamOp       *DeleteStreamOp    `protobuf:"bytes,6,opt,name=deleteStreamOp,proto3" json:"deleteStreamOp,omitempty"`
-	PauseStreamOp        *PauseStreamOp     `protobuf:"bytes,7,opt,name=pauseStreamOp,proto3" json:"pauseStreamOp,omitempty"`
-	ResumeStreamOp       *ResumeStreamOp    `protobuf:"bytes,8,opt,name=resumeStreamOp,proto3" json:"resumeStreamOp,omitempty"`
-	PublishActivityOp    *PublishActivityOp `protobuf:"bytes,9,opt,name=publishActivityOp,proto3" json:"publishActivityOp,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
-	XXX_unrecognized     []byte             `json:"-"`
-	XXX_sizecache        int32              `json:"-"`
+	Op                   Op                   `protobuf:"varint,1,opt,name=op,proto3,enum=protocol.Op" json:"op,omitempty"`
+	CreateStreamOp       *CreateStreamOp      `protobuf:"bytes,2,opt,name=createStreamOp,proto3" json:"createStreamOp,omitempty"`
+	ShrinkISROp          *ShrinkISROp         `protobuf:"bytes,3,opt,name=shrinkISROp,proto3" json:"shrinkISROp,omitempty"`
+	ChangeLeaderOp       *ChangeLeaderOp      `protobuf:"bytes,4,opt,name=changeLeaderOp,proto3" json:"changeLeaderOp,omitempty"`
+	ExpandISROp          *ExpandISROp         `protobuf:"bytes,5,opt,name=expandISROp,proto3" json:"expandISROp,omitempty"`
+	DeleteStreamOp       *DeleteStreamOp      `protobuf:"bytes,6,opt,name=deleteStreamOp,proto3" json:"deleteStreamOp,omitempty"`
+	PauseStreamOp        *PauseStreamOp       `protobuf:"bytes,7,opt,name=pauseStreamOp,proto3" json:"pauseStreamOp,omitempty"`
+	ResumeStreamOp       *ResumeStreamOp      `protobuf:"bytes,8,opt,name=resumeStreamOp,proto3" json:"resumeStreamOp,omitempty"`
+	PublishActivityOp    *PublishActivityOp   `protobuf:"bytes,9,opt,name=publishActivityOp,proto3" json:"publishActivityOp,omitempty"`
+	SetStreamReadonlyOp  *SetStreamReadonlyOp `protobuf:"bytes,10,opt,name=setStreamReadonlyOp,proto3" json:"setStreamReadonlyOp,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}             `json:"-"`
+	XXX_unrecognized     []byte               `json:"-"`
+	XXX_sizecache        int32                `json:"-"`
 }
 
 func (m *RaftLog) Reset()         { *m = RaftLog{} }
@@ -226,6 +230,13 @@ func (m *RaftLog) GetPublishActivityOp() *PublishActivityOp {
 	return nil
 }
 
+func (m *RaftLog) GetSetStreamReadonlyOp() *SetStreamReadonlyOp {
+	if m != nil {
+		return m.SetStreamReadonlyOp
+	}
+	return nil
+}
+
 type CreateStreamOp struct {
 	Stream               *Stream  `protobuf:"bytes,1,opt,name=stream,proto3" json:"stream,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -785,6 +796,69 @@ func (m *PublishActivityOp) GetRaftIndex() uint64 {
 	return 0
 }
 
+type SetStreamReadonlyOp struct {
+	Stream               string   `protobuf:"bytes,1,opt,name=stream,proto3" json:"stream,omitempty"`
+	Partitions           []int32  `protobuf:"varint,2,rep,packed,name=partitions,proto3" json:"partitions,omitempty"`
+	Readonly             bool     `protobuf:"varint,3,opt,name=readonly,proto3" json:"readonly,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *SetStreamReadonlyOp) Reset()         { *m = SetStreamReadonlyOp{} }
+func (m *SetStreamReadonlyOp) String() string { return proto.CompactTextString(m) }
+func (*SetStreamReadonlyOp) ProtoMessage()    {}
+func (*SetStreamReadonlyOp) Descriptor() ([]byte, []int) {
+	return fileDescriptor_7d9410777bf851c3, []int{11}
+}
+func (m *SetStreamReadonlyOp) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *SetStreamReadonlyOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_SetStreamReadonlyOp.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *SetStreamReadonlyOp) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_SetStreamReadonlyOp.Merge(m, src)
+}
+func (m *SetStreamReadonlyOp) XXX_Size() int {
+	return m.Size()
+}
+func (m *SetStreamReadonlyOp) XXX_DiscardUnknown() {
+	xxx_messageInfo_SetStreamReadonlyOp.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_SetStreamReadonlyOp proto.InternalMessageInfo
+
+func (m *SetStreamReadonlyOp) GetStream() string {
+	if m != nil {
+		return m.Stream
+	}
+	return ""
+}
+
+func (m *SetStreamReadonlyOp) GetPartitions() []int32 {
+	if m != nil {
+		return m.Partitions
+	}
+	return nil
+}
+
+func (m *SetStreamReadonlyOp) GetReadonly() bool {
+	if m != nil {
+		return m.Readonly
+	}
+	return false
+}
+
 type NullableInt64 struct {
 	Value                int64    `protobuf:"varint,1,opt,name=value,proto3" json:"value,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -796,7 +870,7 @@ func (m *NullableInt64) Reset()         { *m = NullableInt64{} }
 func (m *NullableInt64) String() string { return proto.CompactTextString(m) }
 func (*NullableInt64) ProtoMessage()    {}
 func (*NullableInt64) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{11}
+	return fileDescriptor_7d9410777bf851c3, []int{12}
 }
 func (m *NullableInt64) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -843,7 +917,7 @@ func (m *NullableInt32) Reset()         { *m = NullableInt32{} }
 func (m *NullableInt32) String() string { return proto.CompactTextString(m) }
 func (*NullableInt32) ProtoMessage()    {}
 func (*NullableInt32) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{12}
+	return fileDescriptor_7d9410777bf851c3, []int{13}
 }
 func (m *NullableInt32) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -890,7 +964,7 @@ func (m *NullableBool) Reset()         { *m = NullableBool{} }
 func (m *NullableBool) String() string { return proto.CompactTextString(m) }
 func (*NullableBool) ProtoMessage()    {}
 func (*NullableBool) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{13}
+	return fileDescriptor_7d9410777bf851c3, []int{14}
 }
 func (m *NullableBool) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -946,7 +1020,7 @@ func (m *StreamConfig) Reset()         { *m = StreamConfig{} }
 func (m *StreamConfig) String() string { return proto.CompactTextString(m) }
 func (*StreamConfig) ProtoMessage()    {}
 func (*StreamConfig) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{14}
+	return fileDescriptor_7d9410777bf851c3, []int{15}
 }
 func (m *StreamConfig) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1059,7 +1133,7 @@ func (m *Stream) Reset()         { *m = Stream{} }
 func (m *Stream) String() string { return proto.CompactTextString(m) }
 func (*Stream) ProtoMessage()    {}
 func (*Stream) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{15}
+	return fileDescriptor_7d9410777bf851c3, []int{16}
 }
 func (m *Stream) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1137,7 +1211,7 @@ func (m *Partition) Reset()         { *m = Partition{} }
 func (m *Partition) String() string { return proto.CompactTextString(m) }
 func (*Partition) ProtoMessage()    {}
 func (*Partition) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{16}
+	return fileDescriptor_7d9410777bf851c3, []int{17}
 }
 func (m *Partition) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1256,7 +1330,7 @@ func (m *RaftJoinRequest) Reset()         { *m = RaftJoinRequest{} }
 func (m *RaftJoinRequest) String() string { return proto.CompactTextString(m) }
 func (*RaftJoinRequest) ProtoMessage()    {}
 func (*RaftJoinRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{17}
+	return fileDescriptor_7d9410777bf851c3, []int{18}
 }
 func (m *RaftJoinRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1311,7 +1385,7 @@ func (m *RaftJoinResponse) Reset()         { *m = RaftJoinResponse{} }
 func (m *RaftJoinResponse) String() string { return proto.CompactTextString(m) }
 func (*RaftJoinResponse) ProtoMessage()    {}
 func (*RaftJoinResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{18}
+	return fileDescriptor_7d9410777bf851c3, []int{19}
 }
 func (m *RaftJoinResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1358,7 +1432,7 @@ func (m *MetadataSnapshot) Reset()         { *m = MetadataSnapshot{} }
 func (m *MetadataSnapshot) String() string { return proto.CompactTextString(m) }
 func (*MetadataSnapshot) ProtoMessage()    {}
 func (*MetadataSnapshot) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{19}
+	return fileDescriptor_7d9410777bf851c3, []int{20}
 }
 func (m *MetadataSnapshot) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1407,7 +1481,7 @@ func (m *ReplicationRequest) Reset()         { *m = ReplicationRequest{} }
 func (m *ReplicationRequest) String() string { return proto.CompactTextString(m) }
 func (*ReplicationRequest) ProtoMessage()    {}
 func (*ReplicationRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{20}
+	return fileDescriptor_7d9410777bf851c3, []int{21}
 }
 func (m *ReplicationRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1468,7 +1542,7 @@ func (m *LeaderEpochOffsetRequest) Reset()         { *m = LeaderEpochOffsetReque
 func (m *LeaderEpochOffsetRequest) String() string { return proto.CompactTextString(m) }
 func (*LeaderEpochOffsetRequest) ProtoMessage()    {}
 func (*LeaderEpochOffsetRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{21}
+	return fileDescriptor_7d9410777bf851c3, []int{22}
 }
 func (m *LeaderEpochOffsetRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1515,7 +1589,7 @@ func (m *LeaderEpochOffsetResponse) Reset()         { *m = LeaderEpochOffsetResp
 func (m *LeaderEpochOffsetResponse) String() string { return proto.CompactTextString(m) }
 func (*LeaderEpochOffsetResponse) ProtoMessage()    {}
 func (*LeaderEpochOffsetResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{22}
+	return fileDescriptor_7d9410777bf851c3, []int{23}
 }
 func (m *LeaderEpochOffsetResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1552,24 +1626,25 @@ func (m *LeaderEpochOffsetResponse) GetEndOffset() int64 {
 }
 
 type PropagatedRequest struct {
-	Op                   Op              `protobuf:"varint,1,opt,name=op,proto3,enum=protocol.Op" json:"op,omitempty"`
-	CreateStreamOp       *CreateStreamOp `protobuf:"bytes,2,opt,name=createStreamOp,proto3" json:"createStreamOp,omitempty"`
-	ShrinkISROp          *ShrinkISROp    `protobuf:"bytes,3,opt,name=shrinkISROp,proto3" json:"shrinkISROp,omitempty"`
-	ReportLeaderOp       *ReportLeaderOp `protobuf:"bytes,4,opt,name=reportLeaderOp,proto3" json:"reportLeaderOp,omitempty"`
-	ExpandISROp          *ExpandISROp    `protobuf:"bytes,5,opt,name=expandISROp,proto3" json:"expandISROp,omitempty"`
-	DeleteStreamOp       *DeleteStreamOp `protobuf:"bytes,6,opt,name=deleteStreamOp,proto3" json:"deleteStreamOp,omitempty"`
-	PauseStreamOp        *PauseStreamOp  `protobuf:"bytes,7,opt,name=pauseStreamOp,proto3" json:"pauseStreamOp,omitempty"`
-	ResumeStreamOp       *ResumeStreamOp `protobuf:"bytes,8,opt,name=resumeStreamOp,proto3" json:"resumeStreamOp,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}        `json:"-"`
-	XXX_unrecognized     []byte          `json:"-"`
-	XXX_sizecache        int32           `json:"-"`
+	Op                   Op                   `protobuf:"varint,1,opt,name=op,proto3,enum=protocol.Op" json:"op,omitempty"`
+	CreateStreamOp       *CreateStreamOp      `protobuf:"bytes,2,opt,name=createStreamOp,proto3" json:"createStreamOp,omitempty"`
+	ShrinkISROp          *ShrinkISROp         `protobuf:"bytes,3,opt,name=shrinkISROp,proto3" json:"shrinkISROp,omitempty"`
+	ReportLeaderOp       *ReportLeaderOp      `protobuf:"bytes,4,opt,name=reportLeaderOp,proto3" json:"reportLeaderOp,omitempty"`
+	ExpandISROp          *ExpandISROp         `protobuf:"bytes,5,opt,name=expandISROp,proto3" json:"expandISROp,omitempty"`
+	DeleteStreamOp       *DeleteStreamOp      `protobuf:"bytes,6,opt,name=deleteStreamOp,proto3" json:"deleteStreamOp,omitempty"`
+	PauseStreamOp        *PauseStreamOp       `protobuf:"bytes,7,opt,name=pauseStreamOp,proto3" json:"pauseStreamOp,omitempty"`
+	ResumeStreamOp       *ResumeStreamOp      `protobuf:"bytes,8,opt,name=resumeStreamOp,proto3" json:"resumeStreamOp,omitempty"`
+	SetStreamReadonlyOp  *SetStreamReadonlyOp `protobuf:"bytes,9,opt,name=setStreamReadonlyOp,proto3" json:"setStreamReadonlyOp,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}             `json:"-"`
+	XXX_unrecognized     []byte               `json:"-"`
+	XXX_sizecache        int32                `json:"-"`
 }
 
 func (m *PropagatedRequest) Reset()         { *m = PropagatedRequest{} }
 func (m *PropagatedRequest) String() string { return proto.CompactTextString(m) }
 func (*PropagatedRequest) ProtoMessage()    {}
 func (*PropagatedRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{23}
+	return fileDescriptor_7d9410777bf851c3, []int{24}
 }
 func (m *PropagatedRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1654,6 +1729,13 @@ func (m *PropagatedRequest) GetResumeStreamOp() *ResumeStreamOp {
 	return nil
 }
 
+func (m *PropagatedRequest) GetSetStreamReadonlyOp() *SetStreamReadonlyOp {
+	if m != nil {
+		return m.SetStreamReadonlyOp
+	}
+	return nil
+}
+
 type Error struct {
 	Code                 uint32   `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"`
 	Msg                  string   `protobuf:"bytes,2,opt,name=msg,proto3" json:"msg,omitempty"`
@@ -1666,7 +1748,7 @@ func (m *Error) Reset()         { *m = Error{} }
 func (m *Error) String() string { return proto.CompactTextString(m) }
 func (*Error) ProtoMessage()    {}
 func (*Error) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{24}
+	return fileDescriptor_7d9410777bf851c3, []int{25}
 }
 func (m *Error) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1721,7 +1803,7 @@ func (m *PropagatedResponse) Reset()         { *m = PropagatedResponse{} }
 func (m *PropagatedResponse) String() string { return proto.CompactTextString(m) }
 func (*PropagatedResponse) ProtoMessage()    {}
 func (*PropagatedResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{25}
+	return fileDescriptor_7d9410777bf851c3, []int{26}
 }
 func (m *PropagatedResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1775,7 +1857,7 @@ func (m *ServerInfoRequest) Reset()         { *m = ServerInfoRequest{} }
 func (m *ServerInfoRequest) String() string { return proto.CompactTextString(m) }
 func (*ServerInfoRequest) ProtoMessage()    {}
 func (*ServerInfoRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{26}
+	return fileDescriptor_7d9410777bf851c3, []int{27}
 }
 func (m *ServerInfoRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1824,7 +1906,7 @@ func (m *ServerInfoResponse) Reset()         { *m = ServerInfoResponse{} }
 func (m *ServerInfoResponse) String() string { return proto.CompactTextString(m) }
 func (*ServerInfoResponse) ProtoMessage()    {}
 func (*ServerInfoResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{27}
+	return fileDescriptor_7d9410777bf851c3, []int{28}
 }
 func (m *ServerInfoResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1886,7 +1968,7 @@ func (m *PartitionStatusRequest) Reset()         { *m = PartitionStatusRequest{}
 func (m *PartitionStatusRequest) String() string { return proto.CompactTextString(m) }
 func (*PartitionStatusRequest) ProtoMessage()    {}
 func (*PartitionStatusRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{28}
+	return fileDescriptor_7d9410777bf851c3, []int{29}
 }
 func (m *PartitionStatusRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1941,7 +2023,7 @@ func (m *PartitionStatusResponse) Reset()         { *m = PartitionStatusResponse
 func (m *PartitionStatusResponse) String() string { return proto.CompactTextString(m) }
 func (*PartitionStatusResponse) ProtoMessage()    {}
 func (*PartitionStatusResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{29}
+	return fileDescriptor_7d9410777bf851c3, []int{30}
 }
 func (m *PartitionStatusResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1996,7 +2078,7 @@ func (m *PartitionNotification) Reset()         { *m = PartitionNotification{} }
 func (m *PartitionNotification) String() string { return proto.CompactTextString(m) }
 func (*PartitionNotification) ProtoMessage()    {}
 func (*PartitionNotification) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{30}
+	return fileDescriptor_7d9410777bf851c3, []int{31}
 }
 func (m *PartitionNotification) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -2052,6 +2134,7 @@ func init() {
 	proto.RegisterType((*ReportLeaderOp)(nil), "protocol.ReportLeaderOp")
 	proto.RegisterType((*ChangeLeaderOp)(nil), "protocol.ChangeLeaderOp")
 	proto.RegisterType((*PublishActivityOp)(nil), "protocol.PublishActivityOp")
+	proto.RegisterType((*SetStreamReadonlyOp)(nil), "protocol.SetStreamReadonlyOp")
 	proto.RegisterType((*NullableInt64)(nil), "protocol.NullableInt64")
 	proto.RegisterType((*NullableInt32)(nil), "protocol.NullableInt32")
 	proto.RegisterType((*NullableBool)(nil), "protocol.NullableBool")
@@ -2077,97 +2160,101 @@ func init() {
 func init() { proto.RegisterFile("server/protocol/internal.proto", fileDescriptor_7d9410777bf851c3) }
 
 var fileDescriptor_7d9410777bf851c3 = []byte{
-	// 1431 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x57, 0xdd, 0x6e, 0x1b, 0xc5,
-	0x17, 0xef, 0xda, 0xb1, 0x63, 0x1f, 0x27, 0x8e, 0x3d, 0xff, 0x36, 0xf5, 0x1f, 0x4a, 0x14, 0x2d,
-	0x54, 0x32, 0x15, 0xa4, 0x22, 0x45, 0x20, 0x10, 0x54, 0x6c, 0x92, 0xa5, 0x31, 0xb5, 0x63, 0x6b,
-	0xd6, 0x45, 0x54, 0x42, 0x44, 0x1b, 0xef, 0xd8, 0x59, 0x58, 0xef, 0x2c, 0x3b, 0xe3, 0x28, 0x7d,
-	0x13, 0x04, 0x57, 0x5c, 0xf5, 0xa2, 0x2f, 0xc2, 0x25, 0x17, 0x3c, 0x00, 0x2a, 0x77, 0x3c, 0x05,
-	0x9a, 0xd9, 0xd9, 0x4f, 0x27, 0x41, 0xa4, 0x37, 0x48, 0xdc, 0xcd, 0x39, 0xf3, 0x3b, 0x1f, 0x73,
-	0xf6, 0xcc, 0xef, 0xcc, 0xc2, 0x16, 0x23, 0xe1, 0x19, 0x09, 0xef, 0x07, 0x21, 0xe5, 0x74, 0x42,
-	0xbd, 0xfb, 0xae, 0xcf, 0x49, 0xe8, 0xdb, 0xde, 0x8e, 0xd4, 0xa0, 0x5a, 0xbc, 0xa1, 0xbf, 0x0d,
-	0x0d, 0x4b, 0x62, 0x2d, 0x6e, 0x73, 0x82, 0x5e, 0x83, 0x5a, 0x64, 0xda, 0x3b, 0xe8, 0x68, 0xdb,
-	0x5a, 0xb7, 0x8e, 0x13, 0x59, 0x7f, 0xb1, 0x02, 0xab, 0xd8, 0x9e, 0xf2, 0x3e, 0x9d, 0xa1, 0x3b,
-	0x50, 0xa2, 0x81, 0x44, 0x34, 0x77, 0xd7, 0x76, 0x62, 0x6f, 0x3b, 0xc3, 0x00, 0x97, 0x68, 0x80,
-	0x3e, 0x83, 0xe6, 0x24, 0x24, 0x36, 0x27, 0x16, 0x0f, 0x89, 0x3d, 0x1f, 0x06, 0x9d, 0xd2, 0xb6,
-	0xd6, 0x6d, 0xec, 0x76, 0x52, 0xe4, 0x7e, 0x6e, 0x1f, 0x17, 0xf0, 0xe8, 0x43, 0x68, 0xb0, 0xd3,
-	0xd0, 0xf5, 0xbf, 0xeb, 0x59, 0x78, 0x18, 0x74, 0xca, 0xd2, 0xfc, 0x56, 0x6a, 0x6e, 0xa5, 0x9b,
-	0x38, 0x8b, 0x94, 0xa1, 0x4f, 0x6d, 0x7f, 0x46, 0xfa, 0xc4, 0x76, 0x48, 0x38, 0x0c, 0x3a, 0x2b,
-	0x4b, 0xa1, 0x73, 0xfb, 0xb8, 0x80, 0x17, 0xa1, 0xc9, 0x79, 0x60, 0xfb, 0x4e, 0x14, 0xba, 0x52,
-	0x0c, 0x6d, 0xa6, 0x9b, 0x38, 0x8b, 0x14, 0xa1, 0x1d, 0xe2, 0x91, 0xcc, 0xa9, 0xab, 0xc5, 0xd0,
-	0x07, 0xb9, 0x7d, 0x5c, 0xc0, 0xa3, 0x4f, 0x61, 0x3d, 0xb0, 0x17, 0x2c, 0x75, 0xb0, 0x2a, 0x1d,
-	0xdc, 0x4e, 0x1d, 0x8c, 0xb2, 0xdb, 0x38, 0x8f, 0x16, 0x09, 0x84, 0x84, 0x2d, 0xe6, 0xa9, 0x7d,
-	0xad, 0x98, 0x00, 0xce, 0xed, 0xe3, 0x02, 0x1e, 0xf5, 0xa0, 0x1d, 0x2c, 0x4e, 0x3c, 0x97, 0x9d,
-	0x1a, 0x13, 0xee, 0x9e, 0xb9, 0xfc, 0xd9, 0x30, 0xe8, 0xd4, 0xa5, 0x93, 0xd7, 0x33, 0x49, 0x14,
-	0x21, 0x78, 0xd9, 0x4a, 0xff, 0x18, 0x9a, 0xf9, 0x6f, 0x8c, 0xba, 0x50, 0x65, 0x72, 0x2d, 0xfb,
-	0xa6, 0xb1, 0xdb, 0xca, 0x7c, 0x4e, 0xa9, 0xc7, 0x6a, 0x5f, 0x7f, 0xae, 0x41, 0x23, 0xf3, 0x85,
-	0xd1, 0x66, 0xce, 0xb2, 0x1e, 0xe3, 0xd0, 0x1d, 0xa8, 0x07, 0x76, 0xc8, 0x5d, 0xee, 0x52, 0x5f,
-	0xb6, 0x58, 0x05, 0xa7, 0x0a, 0xd4, 0x85, 0x8d, 0x90, 0x04, 0x9e, 0x3b, 0xb1, 0xc7, 0x14, 0x93,
-	0x39, 0x3d, 0x23, 0xb2, 0x8f, 0xea, 0xb8, 0xa8, 0x16, 0xfe, 0x3d, 0xf9, 0xf9, 0x65, 0xb3, 0xd4,
-	0xb1, 0x92, 0xd0, 0x36, 0x34, 0xa2, 0x95, 0x19, 0xd0, 0xc9, 0xa9, 0x6c, 0x85, 0x15, 0x9c, 0x55,
-	0xe9, 0x3f, 0x6b, 0xd0, 0xc8, 0x34, 0xc4, 0x35, 0x33, 0xd5, 0x61, 0x2d, 0x49, 0xc9, 0x70, 0x1c,
-	0x95, 0x66, 0x4e, 0xf7, 0x0a, 0x39, 0x76, 0xa1, 0x99, 0xef, 0xbb, 0xcb, 0xb2, 0xd4, 0x09, 0xac,
-	0xe7, 0x1a, 0xec, 0xd2, 0xe3, 0x6c, 0x01, 0x24, 0xd9, 0xb3, 0x4e, 0x69, 0xbb, 0xdc, 0xad, 0xe0,
-	0x8c, 0x46, 0x1c, 0x37, 0xea, 0x2c, 0xc3, 0xf3, 0xe4, 0x69, 0x6a, 0x38, 0x55, 0xe8, 0x87, 0xd0,
-	0xcc, 0xf7, 0xe1, 0x75, 0xe3, 0xe8, 0x3f, 0x6a, 0xc2, 0x55, 0x40, 0x43, 0x9e, 0x5c, 0xdf, 0xeb,
-	0x7d, 0x81, 0x0e, 0xac, 0xaa, 0x6a, 0xab, 0xe2, 0xc7, 0xe2, 0x2b, 0xd4, 0xfd, 0x1b, 0x68, 0xe6,
-	0xa9, 0xe6, 0x9a, 0xb9, 0xa5, 0x19, 0x94, 0xb3, 0x19, 0xe8, 0xef, 0x41, 0x7b, 0xe9, 0x26, 0xca,
-	0xca, 0xdb, 0x53, 0xde, 0xf3, 0x1d, 0x72, 0x2e, 0xa3, 0xac, 0xe0, 0x54, 0xa1, 0xdf, 0x85, 0xf5,
-	0xa3, 0x85, 0xe7, 0xd9, 0x27, 0x1e, 0xe9, 0xf9, 0xfc, 0x83, 0xf7, 0xd1, 0x4d, 0xa8, 0x9c, 0xd9,
-	0xde, 0x82, 0x48, 0x68, 0x19, 0x47, 0x42, 0x01, 0xf6, 0x60, 0x37, 0x0f, 0xab, 0xc4, 0xb0, 0xb7,
-	0x60, 0x2d, 0x86, 0xed, 0x51, 0xea, 0xe5, 0x51, 0xb5, 0x18, 0xf5, 0x5b, 0x05, 0xd6, 0xa2, 0x0f,
-	0xbd, 0x4f, 0xfd, 0xa9, 0x3b, 0x43, 0x26, 0xb4, 0x31, 0xe1, 0xc4, 0x17, 0x87, 0x1b, 0xd8, 0xe7,
-	0x7b, 0xcf, 0x38, 0x61, 0x8a, 0x12, 0x32, 0x4c, 0x97, 0xcb, 0x13, 0x2f, 0x5b, 0xa0, 0xc7, 0x70,
-	0x33, 0xab, 0x1c, 0x10, 0xc6, 0xec, 0x19, 0x61, 0x6a, 0xd4, 0x5c, 0xea, 0xe9, 0x42, 0x23, 0x64,
-	0xc0, 0x46, 0x56, 0x6f, 0xcc, 0x88, 0x9a, 0x39, 0x97, 0xfa, 0x29, 0xe2, 0x85, 0x8b, 0x7d, 0x8f,
-	0xd8, 0x3e, 0x09, 0x7b, 0x62, 0xd8, 0x9e, 0xd9, 0x9e, 0x1a, 0x3d, 0x97, 0xbb, 0x28, 0xe0, 0x85,
-	0x0b, 0x8b, 0xcc, 0xe6, 0xc4, 0xe7, 0x49, 0x5d, 0x2a, 0x7f, 0xe3, 0xa2, 0x80, 0x17, 0x23, 0x24,
-	0x55, 0x89, 0x63, 0x54, 0xaf, 0x76, 0x90, 0x47, 0x8b, 0xa2, 0xee, 0xd3, 0x79, 0x60, 0x4f, 0x84,
-	0xe2, 0x11, 0x0d, 0xe9, 0x82, 0xbb, 0x3e, 0x61, 0xcb, 0x83, 0x28, 0xd7, 0x1f, 0xf8, 0x42, 0x23,
-	0xf4, 0x10, 0x9a, 0x4a, 0x6f, 0xfa, 0x02, 0xeb, 0xa8, 0x79, 0xb4, 0xb9, 0xec, 0x46, 0xf4, 0x0f,
-	0x2e, 0xa0, 0xc5, 0x59, 0x8c, 0x05, 0xa7, 0x92, 0x92, 0xc6, 0xee, 0x9c, 0xa8, 0x49, 0x74, 0xf9,
-	0x59, 0x72, 0x68, 0xf4, 0x35, 0xbc, 0x91, 0x28, 0x0e, 0x5c, 0x26, 0x71, 0x53, 0x6b, 0x71, 0xc2,
-	0x26, 0xa1, 0x7b, 0x42, 0x42, 0xd6, 0x81, 0x2b, 0xb3, 0xb9, 0xda, 0x58, 0xff, 0x49, 0x83, 0x6a,
-	0xd4, 0xd6, 0x08, 0xc1, 0x8a, 0x6f, 0xcf, 0x89, 0xba, 0xd4, 0x72, 0x2d, 0x08, 0x85, 0x2d, 0x4e,
-	0xbe, 0x25, 0x13, 0x2e, 0x1b, 0xb2, 0x8e, 0x63, 0x11, 0x3d, 0xc8, 0x71, 0x5a, 0x79, 0xbb, 0xdc,
-	0x6d, 0xec, 0xfe, 0x2f, 0x3b, 0xe1, 0xd5, 0x5e, 0x8e, 0x50, 0x77, 0xa0, 0x3a, 0x91, 0xb7, 0x47,
-	0xf5, 0xd4, 0x66, 0x71, 0x76, 0x46, 0x77, 0x0b, 0x2b, 0x94, 0xfe, 0xbc, 0x04, 0xf5, 0x51, 0x96,
-	0xdd, 0xe2, 0x64, 0xb4, 0x7c, 0x32, 0x29, 0x23, 0x95, 0x72, 0x8c, 0xd4, 0x84, 0x92, 0x1b, 0xcd,
-	0xa1, 0x0a, 0x2e, 0xb9, 0x8e, 0xb8, 0xda, 0xb3, 0x90, 0x2e, 0x02, 0x45, 0x82, 0x91, 0x80, 0xde,
-	0x81, 0xb6, 0xa2, 0x49, 0x11, 0xe6, 0x73, 0x7b, 0xc2, 0x69, 0x28, 0x3b, 0xb6, 0x82, 0x97, 0x37,
-	0xc4, 0xdb, 0x52, 0x29, 0x59, 0xa7, 0xba, 0x5d, 0x16, 0x6f, 0xcb, 0x58, 0xce, 0x70, 0xdc, 0x6a,
-	0x8e, 0x65, 0x5b, 0x50, 0x76, 0x59, 0xd8, 0xa9, 0x49, 0xb8, 0x58, 0x16, 0x79, 0xb7, 0xbe, 0xc4,
-	0xbb, 0x22, 0x57, 0x22, 0xf7, 0x40, 0xee, 0x45, 0x82, 0x88, 0x20, 0x5f, 0x4b, 0x4e, 0xa7, 0x21,
-	0xd9, 0x49, 0x49, 0xba, 0x09, 0x1b, 0xe2, 0x51, 0xfb, 0x05, 0x75, 0x7d, 0x4c, 0xbe, 0x5f, 0x10,
-	0x26, 0x8b, 0xe2, 0x53, 0x87, 0x24, 0x4f, 0x60, 0x25, 0x89, 0x03, 0x88, 0x95, 0xe1, 0x38, 0xa1,
-	0x2a, 0x57, 0x22, 0xeb, 0x5d, 0x68, 0xa5, 0x6e, 0x58, 0x40, 0x7d, 0x46, 0x64, 0x22, 0x61, 0x48,
-	0x43, 0xe5, 0x26, 0x12, 0xf4, 0x87, 0xd0, 0x1a, 0x10, 0x6e, 0x3b, 0x36, 0xb7, 0x2d, 0xdf, 0x0e,
-	0xd8, 0x29, 0xe5, 0xe8, 0x1e, 0xac, 0x46, 0x85, 0x17, 0x44, 0x58, 0xbe, 0xf0, 0x6d, 0x14, 0x03,
-	0x74, 0x0f, 0x10, 0x4e, 0x6b, 0x1b, 0xe7, 0x2c, 0x27, 0xae, 0xd4, 0x26, 0x69, 0xa7, 0x0a, 0x71,
-	0x22, 0x3a, 0x9d, 0x32, 0x12, 0x35, 0x63, 0x19, 0x2b, 0xa9, 0x58, 0xcc, 0xf2, 0xf2, 0x10, 0xfb,
-	0x04, 0x3a, 0xfd, 0x54, 0x1c, 0x4a, 0xb3, 0x38, 0x66, 0xc1, 0x5a, 0x5b, 0xb6, 0xfe, 0x08, 0xfe,
-	0x7f, 0x81, 0xb5, 0x2a, 0xcf, 0x1d, 0xa8, 0x13, 0xdf, 0x89, 0x94, 0x6a, 0xfe, 0xa4, 0x0a, 0xfd,
-	0xcf, 0x32, 0xb4, 0x47, 0x21, 0x0d, 0xec, 0x99, 0xcd, 0x89, 0x93, 0x1e, 0xf3, 0xdf, 0xfb, 0xdf,
-	0x11, 0xe6, 0x1e, 0x22, 0xcb, 0xff, 0x1d, 0xf9, 0x87, 0x0a, 0x2e, 0xe0, 0xff, 0xcb, 0xff, 0x1d,
-	0xfa, 0xbb, 0x50, 0x31, 0xc5, 0xe5, 0x10, 0x54, 0x3a, 0xa1, 0x4e, 0x44, 0xa5, 0xeb, 0x58, 0xae,
-	0x05, 0x07, 0xcc, 0xd9, 0x4c, 0xdd, 0x38, 0xb1, 0xd4, 0x9f, 0x02, 0xca, 0xb6, 0x46, 0xd2, 0x4f,
-	0x57, 0xf5, 0xc6, 0xdd, 0xf8, 0x32, 0x46, 0x2d, 0xb1, 0x91, 0x29, 0xac, 0x50, 0xc7, 0xb7, 0xf3,
-	0x4d, 0x68, 0x47, 0xff, 0xc3, 0x3d, 0x7f, 0x4a, 0xe3, 0xae, 0x8b, 0xd8, 0x30, 0xba, 0x55, 0x25,
-	0xd7, 0xd1, 0xfb, 0x80, 0xb2, 0x20, 0x15, 0xbf, 0x80, 0x12, 0x67, 0x39, 0xa5, 0x2c, 0xe6, 0x7f,
-	0xb9, 0x16, 0x3a, 0xf1, 0xd1, 0x15, 0xb3, 0xca, 0xb5, 0x7e, 0x04, 0x9b, 0x09, 0x55, 0x8b, 0xbf,
-	0xf0, 0x05, 0xcb, 0x10, 0xd1, 0x3f, 0x7f, 0x2f, 0xea, 0x03, 0xb8, 0xbd, 0xe4, 0x4f, 0xa5, 0xb8,
-	0x09, 0x55, 0x72, 0xee, 0x32, 0xce, 0xd4, 0x13, 0x4d, 0x49, 0x82, 0xd9, 0x5c, 0x16, 0x75, 0xa2,
-	0xf4, 0x57, 0xc3, 0x89, 0xac, 0x0f, 0xe0, 0x56, 0xe2, 0xee, 0x88, 0x72, 0x77, 0xaa, 0x98, 0xe7,
-	0x7a, 0xd9, 0xdd, 0x7b, 0xa1, 0x41, 0x69, 0x18, 0xa0, 0x36, 0xac, 0xef, 0x63, 0xd3, 0x18, 0x9b,
-	0xc7, 0xd6, 0x18, 0x9b, 0xc6, 0xa0, 0x75, 0x03, 0x35, 0x01, 0xac, 0x43, 0xdc, 0x3b, 0x7a, 0x7c,
-	0xdc, 0xb3, 0x70, 0x4b, 0x13, 0x10, 0x6c, 0x8e, 0x86, 0x78, 0x7c, 0xdc, 0x37, 0x8d, 0x03, 0x13,
-	0xb7, 0x4a, 0xd2, 0xea, 0xd0, 0x38, 0x7a, 0x64, 0xc6, 0xaa, 0xb2, 0xb0, 0x32, 0xbf, 0x1a, 0x19,
-	0x47, 0x07, 0xd2, 0x6a, 0x45, 0x40, 0x0e, 0xcc, 0xbe, 0x99, 0x3a, 0xae, 0xa0, 0x16, 0xac, 0x8d,
-	0x8c, 0x27, 0x56, 0xa2, 0xa9, 0x46, 0xae, 0xad, 0x27, 0x83, 0x44, 0xb5, 0x8a, 0x6e, 0x42, 0x6b,
-	0xf4, 0x64, 0xaf, 0xdf, 0xb3, 0x0e, 0x8f, 0x8d, 0xfd, 0x71, 0xef, 0xcb, 0xde, 0xf8, 0x69, 0xab,
-	0xb6, 0xd7, 0xfa, 0xe5, 0xe5, 0x96, 0xf6, 0xeb, 0xcb, 0x2d, 0xed, 0xf7, 0x97, 0x5b, 0xda, 0x0f,
-	0x7f, 0x6c, 0xdd, 0x38, 0xa9, 0xca, 0xbe, 0x79, 0xf0, 0x57, 0x00, 0x00, 0x00, 0xff, 0xff, 0x40,
-	0x1e, 0x59, 0xee, 0x63, 0x11, 0x00, 0x00,
+	// 1502 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x58, 0x5f, 0x6f, 0x1b, 0xc5,
+	0x16, 0xaf, 0xed, 0xd8, 0xb1, 0x8f, 0x13, 0xc7, 0x99, 0xb4, 0xa9, 0x6f, 0x6f, 0x1b, 0x45, 0x7b,
+	0x6f, 0x25, 0xdf, 0xea, 0xde, 0x54, 0x37, 0x45, 0x20, 0x10, 0x54, 0x6c, 0x92, 0xa5, 0x31, 0x75,
+	0xe2, 0x68, 0xd6, 0x45, 0x54, 0x42, 0x44, 0x13, 0xef, 0xd8, 0x59, 0x58, 0xef, 0x2c, 0x3b, 0xe3,
+	0x28, 0xfd, 0x26, 0x08, 0x9e, 0x78, 0xea, 0x87, 0xe0, 0x95, 0x07, 0x1e, 0x79, 0xe0, 0x03, 0xa0,
+	0xf2, 0x35, 0x78, 0x40, 0x33, 0x3b, 0xfb, 0xd7, 0x71, 0x11, 0xe9, 0x0b, 0x12, 0x6f, 0x73, 0xce,
+	0xfc, 0xce, 0x99, 0x33, 0xe3, 0xdf, 0xfc, 0xce, 0xac, 0x61, 0x8b, 0xd3, 0xf0, 0x82, 0x86, 0x0f,
+	0x83, 0x90, 0x09, 0x36, 0x62, 0xde, 0x43, 0xd7, 0x17, 0x34, 0xf4, 0x89, 0xb7, 0xa3, 0x3c, 0xa8,
+	0x1e, 0x4f, 0x18, 0xff, 0x81, 0xa6, 0xad, 0xb0, 0xb6, 0x20, 0x82, 0xa2, 0x3b, 0x50, 0x8f, 0x42,
+	0x7b, 0x07, 0x9d, 0xd2, 0x76, 0xa9, 0xdb, 0xc0, 0x89, 0x6d, 0xfc, 0xb6, 0x04, 0xcb, 0x98, 0x8c,
+	0x45, 0x9f, 0x4d, 0xd0, 0x5d, 0x28, 0xb3, 0x40, 0x21, 0x5a, 0xbb, 0x2b, 0x3b, 0x71, 0xb6, 0x9d,
+	0x41, 0x80, 0xcb, 0x2c, 0x40, 0x1f, 0x42, 0x6b, 0x14, 0x52, 0x22, 0xa8, 0x2d, 0x42, 0x4a, 0xa6,
+	0x83, 0xa0, 0x53, 0xde, 0x2e, 0x75, 0x9b, 0xbb, 0x9d, 0x14, 0xb9, 0x9f, 0x9b, 0xc7, 0x05, 0x3c,
+	0x7a, 0x07, 0x9a, 0xfc, 0x3c, 0x74, 0xfd, 0x2f, 0x7b, 0x36, 0x1e, 0x04, 0x9d, 0x8a, 0x0a, 0xbf,
+	0x95, 0x86, 0xdb, 0xe9, 0x24, 0xce, 0x22, 0xd5, 0xd2, 0xe7, 0xc4, 0x9f, 0xd0, 0x3e, 0x25, 0x0e,
+	0x0d, 0x07, 0x41, 0x67, 0x69, 0x6e, 0xe9, 0xdc, 0x3c, 0x2e, 0xe0, 0xe5, 0xd2, 0xf4, 0x32, 0x20,
+	0xbe, 0x13, 0x2d, 0x5d, 0x2d, 0x2e, 0x6d, 0xa5, 0x93, 0x38, 0x8b, 0x94, 0x4b, 0x3b, 0xd4, 0xa3,
+	0x99, 0x5d, 0xd7, 0x8a, 0x4b, 0x1f, 0xe4, 0xe6, 0x71, 0x01, 0x8f, 0x3e, 0x80, 0xd5, 0x80, 0xcc,
+	0x78, 0x9a, 0x60, 0x59, 0x25, 0xb8, 0x9d, 0x26, 0x38, 0xc9, 0x4e, 0xe3, 0x3c, 0x5a, 0x16, 0x10,
+	0x52, 0x3e, 0x9b, 0xa6, 0xf1, 0xf5, 0x62, 0x01, 0x38, 0x37, 0x8f, 0x0b, 0x78, 0xd4, 0x83, 0xf5,
+	0x60, 0x76, 0xe6, 0xb9, 0xfc, 0xdc, 0x1c, 0x09, 0xf7, 0xc2, 0x15, 0x2f, 0x06, 0x41, 0xa7, 0xa1,
+	0x92, 0xfc, 0x33, 0x53, 0x44, 0x11, 0x82, 0xe7, 0xa3, 0xd0, 0x00, 0x36, 0x38, 0x15, 0x51, 0x66,
+	0x4c, 0x89, 0xc3, 0x7c, 0x4f, 0x26, 0x03, 0x95, 0xec, 0x5e, 0xe6, 0x97, 0x9c, 0x07, 0xe1, 0xab,
+	0x22, 0x8d, 0xf7, 0xa0, 0x95, 0x27, 0x0d, 0xea, 0x42, 0x8d, 0xab, 0xb1, 0x22, 0x62, 0x73, 0xb7,
+	0x9d, 0xc9, 0x1a, 0x45, 0xeb, 0x79, 0xe3, 0x65, 0x09, 0x9a, 0x19, 0xca, 0xa0, 0xcd, 0x5c, 0x64,
+	0x23, 0xc6, 0xa1, 0xbb, 0xd0, 0x08, 0x48, 0x28, 0x5c, 0xe1, 0x32, 0x5f, 0x71, 0xb6, 0x8a, 0x53,
+	0x07, 0xea, 0xc2, 0x5a, 0x48, 0x03, 0xcf, 0x1d, 0x91, 0x21, 0xc3, 0x74, 0xca, 0x2e, 0xa8, 0x22,
+	0x66, 0x03, 0x17, 0xdd, 0x32, 0xbf, 0xa7, 0xf8, 0xa4, 0xd8, 0xd7, 0xc0, 0xda, 0x42, 0xdb, 0xd0,
+	0x8c, 0x46, 0x56, 0xc0, 0x46, 0xe7, 0x8a, 0x5b, 0x4b, 0x38, 0xeb, 0x32, 0xbe, 0x2b, 0x41, 0x33,
+	0xc3, 0xb0, 0x6b, 0x56, 0x6a, 0xc0, 0x4a, 0x52, 0x92, 0xe9, 0x38, 0xba, 0xcc, 0x9c, 0xef, 0x0d,
+	0x6a, 0xec, 0x42, 0x2b, 0x4f, 0xe4, 0x45, 0x55, 0x1a, 0x14, 0x56, 0x73, 0x8c, 0x5d, 0xb8, 0x9d,
+	0x2d, 0x80, 0xa4, 0x7a, 0xde, 0x29, 0x6f, 0x57, 0xba, 0x55, 0x9c, 0xf1, 0xc8, 0xed, 0x46, 0x54,
+	0x35, 0x3d, 0x4f, 0xed, 0xa6, 0x8e, 0x53, 0x87, 0x71, 0x08, 0xad, 0x3c, 0xb1, 0xaf, 0xbb, 0x8e,
+	0xf1, 0x4d, 0x49, 0xa6, 0x0a, 0x58, 0x28, 0x12, 0x3d, 0xb8, 0xde, 0x2f, 0xd0, 0x81, 0x65, 0x7d,
+	0xda, 0xfa, 0xf0, 0x63, 0xf3, 0x0d, 0xce, 0xfd, 0x73, 0x68, 0xe5, 0xb5, 0xeb, 0x9a, 0xb5, 0xa5,
+	0x15, 0x54, 0xb2, 0x15, 0x18, 0xff, 0x87, 0xf5, 0xb9, 0xab, 0xad, 0x4e, 0x9e, 0x8c, 0x45, 0xcf,
+	0x77, 0xe8, 0xa5, 0x5a, 0x65, 0x09, 0xa7, 0x0e, 0xc3, 0x85, 0x8d, 0x2b, 0x2e, 0xf0, 0xb5, 0x7f,
+	0xe6, 0x3b, 0x50, 0x0f, 0x75, 0x16, 0xfd, 0x2b, 0x27, 0xb6, 0x71, 0x1f, 0x56, 0x8f, 0x67, 0x9e,
+	0x47, 0xce, 0x3c, 0xda, 0xf3, 0xc5, 0xdb, 0x6f, 0xa1, 0x9b, 0x50, 0xbd, 0x20, 0xde, 0x8c, 0xaa,
+	0x35, 0x2a, 0x38, 0x32, 0x0a, 0xb0, 0x47, 0xbb, 0x79, 0x58, 0x35, 0x86, 0xfd, 0x1b, 0x56, 0x62,
+	0xd8, 0x1e, 0x63, 0x5e, 0x1e, 0x55, 0x8f, 0x51, 0x3f, 0x57, 0x61, 0x25, 0xda, 0xdc, 0x3e, 0xf3,
+	0xc7, 0xee, 0x04, 0x59, 0xb0, 0x8e, 0xa9, 0xa0, 0xbe, 0x2c, 0xf7, 0x88, 0x5c, 0xee, 0xbd, 0x10,
+	0x94, 0x6b, 0xf5, 0xc9, 0xa8, 0x74, 0xae, 0x4e, 0x3c, 0x1f, 0x81, 0x9e, 0xc2, 0xcd, 0xac, 0xf3,
+	0x88, 0x72, 0x4e, 0x26, 0x94, 0xeb, 0x36, 0xb9, 0x30, 0xd3, 0x95, 0x41, 0xc8, 0x84, 0xb5, 0xac,
+	0xdf, 0x9c, 0x50, 0xdd, 0x2f, 0x17, 0xe6, 0x29, 0xe2, 0x65, 0x8a, 0x7d, 0x8f, 0x12, 0x9f, 0x86,
+	0x3d, 0xf9, 0x50, 0xb8, 0x20, 0x9e, 0x6e, 0x9b, 0x8b, 0x53, 0x14, 0xf0, 0x32, 0x85, 0x4d, 0x27,
+	0x53, 0xea, 0x8b, 0xe4, 0x5c, 0xaa, 0x7f, 0x90, 0xa2, 0x80, 0x97, 0xed, 0x2f, 0x75, 0xc9, 0x6d,
+	0xd4, 0x5e, 0x9f, 0x20, 0x8f, 0x96, 0x87, 0xba, 0xcf, 0xa6, 0x01, 0x19, 0x49, 0xc7, 0x13, 0x16,
+	0xb2, 0x99, 0x70, 0x7d, 0xca, 0xe7, 0x9b, 0x68, 0x8e, 0x1f, 0xf8, 0xca, 0x20, 0xf4, 0x18, 0x5a,
+	0xda, 0x6f, 0xf9, 0x12, 0xeb, 0xe8, 0x5e, 0xba, 0x39, 0x9f, 0x46, 0xf2, 0x07, 0x17, 0xd0, 0x72,
+	0x2f, 0xe6, 0x4c, 0x30, 0xa5, 0x7e, 0x43, 0x77, 0x4a, 0x75, 0x17, 0x5d, 0xbc, 0x97, 0x1c, 0x1a,
+	0x7d, 0x06, 0xf7, 0x12, 0xc7, 0x81, 0xcb, 0x15, 0x6e, 0x6c, 0xcf, 0xce, 0xf8, 0x28, 0x74, 0xcf,
+	0x68, 0xc8, 0x75, 0x1f, 0x5d, 0x54, 0xcd, 0xeb, 0x83, 0x8d, 0x6f, 0x4b, 0x50, 0x8b, 0x68, 0x8d,
+	0x10, 0x2c, 0xf9, 0x64, 0x4a, 0xf5, 0x3d, 0x55, 0x63, 0xa9, 0x5d, 0x7c, 0x76, 0xf6, 0x05, 0x1d,
+	0x09, 0x45, 0xc8, 0x06, 0x8e, 0x4d, 0xf4, 0x28, 0x77, 0x7f, 0x2b, 0xdb, 0x95, 0x6e, 0x73, 0x77,
+	0x23, 0xfb, 0x3a, 0xd1, 0x73, 0xb9, 0x4b, 0xbd, 0x03, 0xb5, 0x91, 0xba, 0x3d, 0x9a, 0x53, 0x9b,
+	0xc5, 0x36, 0x1d, 0xdd, 0x2d, 0xac, 0x51, 0xc6, 0xcb, 0x32, 0x34, 0x4e, 0xb2, 0x42, 0x1a, 0x17,
+	0x53, 0xca, 0x17, 0x93, 0x8a, 0x4c, 0x39, 0x27, 0x32, 0x2d, 0x28, 0xbb, 0x51, 0xcb, 0xab, 0xe2,
+	0xb2, 0xeb, 0xc8, 0xab, 0x3d, 0x09, 0xd9, 0x2c, 0xd0, 0x7a, 0x1b, 0x19, 0xe8, 0xbf, 0xb0, 0xae,
+	0x15, 0x59, 0x2e, 0xf3, 0x11, 0x19, 0x09, 0x16, 0x2a, 0xc6, 0x56, 0xf1, 0xfc, 0x44, 0x24, 0x4c,
+	0xca, 0xc9, 0x3b, 0xb5, 0xed, 0x8a, 0x7c, 0x17, 0xc7, 0x76, 0x46, 0x4e, 0x97, 0x73, 0x82, 0xde,
+	0x86, 0x8a, 0xcb, 0xc3, 0x4e, 0x5d, 0xc1, 0xe5, 0xb0, 0x28, 0xf1, 0x8d, 0x39, 0x89, 0x97, 0xb5,
+	0x52, 0x35, 0x07, 0x6a, 0x2e, 0x32, 0xe4, 0x0a, 0xea, 0xa5, 0xe7, 0x74, 0x9a, 0x4a, 0x9d, 0xb4,
+	0x65, 0x58, 0xb0, 0x26, 0x1f, 0xe4, 0x1f, 0x33, 0xd7, 0xc7, 0xf4, 0xab, 0x19, 0xe5, 0xea, 0x50,
+	0x7c, 0xe6, 0xd0, 0xe4, 0xf9, 0xae, 0x2d, 0xb9, 0x01, 0x39, 0x32, 0x1d, 0x27, 0xd4, 0xc7, 0x95,
+	0xd8, 0x46, 0x17, 0xda, 0x69, 0x1a, 0x1e, 0x30, 0x9f, 0x53, 0x55, 0x48, 0x18, 0xb2, 0x50, 0xa7,
+	0x89, 0x0c, 0xe3, 0x31, 0xb4, 0x8f, 0xa8, 0x20, 0x0e, 0x11, 0xc4, 0xf6, 0x49, 0xc0, 0xcf, 0x99,
+	0x40, 0x0f, 0x60, 0x39, 0x3a, 0x78, 0x29, 0x84, 0x95, 0x2b, 0x9f, 0x61, 0x31, 0xc0, 0xf0, 0x00,
+	0xe1, 0xf4, 0x6c, 0xe3, 0x9a, 0x55, 0x73, 0x57, 0xde, 0xa4, 0xec, 0xd4, 0x21, 0x77, 0xc4, 0xc6,
+	0x63, 0x4e, 0x23, 0x32, 0x56, 0xb0, 0xb6, 0x8a, 0x87, 0x59, 0x99, 0xef, 0x97, 0xef, 0x43, 0xa7,
+	0x9f, 0x9a, 0x03, 0x15, 0x16, 0xaf, 0x59, 0x88, 0x2e, 0xcd, 0x47, 0xbf, 0x0b, 0xff, 0xb8, 0x22,
+	0x5a, 0x1f, 0xcf, 0x5d, 0x68, 0x50, 0xdf, 0x89, 0x9c, 0xba, 0xff, 0xa4, 0x0e, 0xe3, 0xfb, 0x25,
+	0x58, 0x3f, 0x09, 0x59, 0x40, 0x26, 0x44, 0x50, 0x27, 0xdd, 0xe6, 0x5f, 0xf7, 0x9b, 0x29, 0xcc,
+	0xbd, 0x79, 0xe6, 0xbf, 0x99, 0xf2, 0x6f, 0x22, 0x5c, 0xc0, 0xff, 0xad, 0xbf, 0x99, 0x16, 0x7c,
+	0xe8, 0x34, 0xae, 0xfd, 0xa1, 0xf3, 0x3f, 0xa8, 0x5a, 0xf2, 0xb6, 0x49, 0x6d, 0x1e, 0x31, 0x27,
+	0xd2, 0xe6, 0x55, 0xac, 0xc6, 0x52, 0x54, 0xa6, 0x7c, 0xa2, 0xaf, 0xb0, 0x1c, 0x1a, 0xcf, 0x01,
+	0x65, 0xb9, 0x96, 0x10, 0xf4, 0x75, 0x64, 0xbb, 0x1f, 0xdf, 0xee, 0x88, 0x63, 0x6b, 0x99, 0x5f,
+	0x4a, 0xba, 0xe3, 0xeb, 0xfe, 0x2f, 0x58, 0x8f, 0xfe, 0x1c, 0xe8, 0xf9, 0x63, 0x16, 0xd3, 0x38,
+	0x92, 0xd7, 0xe8, 0x9a, 0x96, 0x5d, 0xc7, 0xe8, 0x03, 0xca, 0x82, 0xf4, 0xfa, 0x05, 0x94, 0xdc,
+	0xcb, 0x39, 0xe3, 0x71, 0x43, 0x51, 0x63, 0xe9, 0x93, 0x2c, 0xd2, 0x52, 0xad, 0xc6, 0xc6, 0x31,
+	0x6c, 0x26, 0xda, 0x6f, 0x0b, 0x22, 0x66, 0x3c, 0xa3, 0x6c, 0x7f, 0xfe, 0xad, 0x6b, 0x1c, 0xc1,
+	0xed, 0xb9, 0x7c, 0xba, 0xc4, 0x4d, 0xa8, 0xd1, 0x4b, 0x97, 0x0b, 0xae, 0xdf, 0x7c, 0xda, 0x92,
+	0x52, 0xe9, 0xf2, 0x88, 0xda, 0x2a, 0x5f, 0x1d, 0x27, 0xb6, 0x71, 0x04, 0xb7, 0x92, 0x74, 0xc7,
+	0x4c, 0xb8, 0x63, 0x2d, 0x65, 0xd7, 0xab, 0xee, 0xc1, 0x0f, 0x25, 0x28, 0x0f, 0x02, 0xb4, 0x0e,
+	0xab, 0xfb, 0xd8, 0x32, 0x87, 0xd6, 0xa9, 0x3d, 0xc4, 0x96, 0x79, 0xd4, 0xbe, 0x81, 0x5a, 0x00,
+	0xf6, 0x21, 0xee, 0x1d, 0x3f, 0x3d, 0xed, 0xd9, 0xb8, 0x5d, 0x92, 0x10, 0x6c, 0x9d, 0x0c, 0xf0,
+	0xf0, 0xb4, 0x6f, 0x99, 0x07, 0x16, 0x6e, 0x97, 0x55, 0xd4, 0xa1, 0x79, 0xfc, 0xc4, 0x8a, 0x5d,
+	0x15, 0x19, 0x65, 0x7d, 0x7a, 0x62, 0x1e, 0x1f, 0xa8, 0xa8, 0x25, 0x09, 0x39, 0xb0, 0xfa, 0x56,
+	0x9a, 0xb8, 0x8a, 0xda, 0xb0, 0x72, 0x62, 0x3e, 0xb3, 0x13, 0x4f, 0x2d, 0x4a, 0x6d, 0x3f, 0x3b,
+	0x4a, 0x5c, 0xcb, 0xe8, 0x26, 0xb4, 0x4f, 0x9e, 0xed, 0xf5, 0x7b, 0xf6, 0xe1, 0xa9, 0xb9, 0x3f,
+	0xec, 0x7d, 0xd2, 0x1b, 0x3e, 0x6f, 0xd7, 0xd1, 0x6d, 0xd8, 0xb0, 0xad, 0xa1, 0x46, 0x9d, 0x62,
+	0xcb, 0x3c, 0x18, 0x1c, 0xf7, 0x9f, 0xb7, 0x1b, 0x7b, 0xed, 0x1f, 0x5f, 0x6d, 0x95, 0x7e, 0x7a,
+	0xb5, 0x55, 0xfa, 0xe5, 0xd5, 0x56, 0xe9, 0xeb, 0x5f, 0xb7, 0x6e, 0x9c, 0xd5, 0x14, 0xa1, 0x1e,
+	0xfd, 0x1e, 0x00, 0x00, 0xff, 0xff, 0x87, 0xf0, 0xf5, 0xe6, 0x89, 0x12, 0x00, 0x00,
 }
 
 func (m *ServerState) Marshal() (dAtA []byte, err error) {
@@ -2228,6 +2315,18 @@ func (m *RaftLog) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.SetStreamReadonlyOp != nil {
+		{
+			size, err := m.SetStreamReadonlyOp.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintInternal(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x52
+	}
 	if m.PublishActivityOp != nil {
 		{
 			size, err := m.PublishActivityOp.MarshalToSizedBuffer(dAtA[:i])
@@ -2556,21 +2655,21 @@ func (m *PauseStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		dAtA[i] = 0x18
 	}
 	if len(m.Partitions) > 0 {
-		dAtA11 := make([]byte, len(m.Partitions)*10)
-		var j10 int
+		dAtA12 := make([]byte, len(m.Partitions)*10)
+		var j11 int
 		for _, num1 := range m.Partitions {
 			num := uint64(num1)
 			for num >= 1<<7 {
-				dAtA11[j10] = uint8(uint64(num)&0x7f | 0x80)
+				dAtA12[j11] = uint8(uint64(num)&0x7f | 0x80)
 				num >>= 7
-				j10++
+				j11++
 			}
-			dAtA11[j10] = uint8(num)
-			j10++
+			dAtA12[j11] = uint8(num)
+			j11++
 		}
-		i -= j10
-		copy(dAtA[i:], dAtA11[:j10])
-		i = encodeVarintInternal(dAtA, i, uint64(j10))
+		i -= j11
+		copy(dAtA[i:], dAtA12[:j11])
+		i = encodeVarintInternal(dAtA, i, uint64(j11))
 		i--
 		dAtA[i] = 0x12
 	}
@@ -2609,21 +2708,21 @@ func (m *ResumeStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
 	if len(m.Partitions) > 0 {
-		dAtA13 := make([]byte, len(m.Partitions)*10)
-		var j12 int
+		dAtA14 := make([]byte, len(m.Partitions)*10)
+		var j13 int
 		for _, num1 := range m.Partitions {
 			num := uint64(num1)
 			for num >= 1<<7 {
-				dAtA13[j12] = uint8(uint64(num)&0x7f | 0x80)
+				dAtA14[j13] = uint8(uint64(num)&0x7f | 0x80)
 				num >>= 7
-				j12++
+				j13++
 			}
-			dAtA13[j12] = uint8(num)
-			j12++
+			dAtA14[j13] = uint8(num)
+			j13++
 		}
-		i -= j12
-		copy(dAtA[i:], dAtA13[:j12])
-		i = encodeVarintInternal(dAtA, i, uint64(j12))
+		i -= j13
+		copy(dAtA[i:], dAtA14[:j13])
+		i = encodeVarintInternal(dAtA, i, uint64(j13))
 		i--
 		dAtA[i] = 0x12
 	}
@@ -2773,6 +2872,69 @@ func (m *PublishActivityOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *SetStreamReadonlyOp) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *SetStreamReadonlyOp) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *SetStreamReadonlyOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.Readonly {
+		i--
+		if m.Readonly {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x18
+	}
+	if len(m.Partitions) > 0 {
+		dAtA16 := make([]byte, len(m.Partitions)*10)
+		var j15 int
+		for _, num1 := range m.Partitions {
+			num := uint64(num1)
+			for num >= 1<<7 {
+				dAtA16[j15] = uint8(uint64(num)&0x7f | 0x80)
+				num >>= 7
+				j15++
+			}
+			dAtA16[j15] = uint8(num)
+			j15++
+		}
+		i -= j15
+		copy(dAtA[i:], dAtA16[:j15])
+		i = encodeVarintInternal(dAtA, i, uint64(j15))
+		i--
+		dAtA[i] = 0x12
+	}
+	if len(m.Stream) > 0 {
+		i -= len(m.Stream)
+		copy(dAtA[i:], m.Stream)
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *NullableInt64) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -3439,6 +3601,18 @@ func (m *PropagatedRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.SetStreamReadonlyOp != nil {
+		{
+			size, err := m.SetStreamReadonlyOp.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintInternal(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0x4a
+	}
 	if m.ResumeStreamOp != nil {
 		{
 			size, err := m.ResumeStreamOp.MarshalToSizedBuffer(dAtA[:i])
@@ -3887,6 +4061,10 @@ func (m *RaftLog) Size() (n int) {
 		l = m.PublishActivityOp.Size()
 		n += 1 + l + sovInternal(uint64(l))
 	}
+	if m.SetStreamReadonlyOp != nil {
+		l = m.SetStreamReadonlyOp.Size()
+		n += 1 + l + sovInternal(uint64(l))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -4102,6 +4280,32 @@ func (m *PublishActivityOp) Size() (n int) {
 	return n
 }
 
+func (m *SetStreamReadonlyOp) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = len(m.Stream)
+	if l > 0 {
+		n += 1 + l + sovInternal(uint64(l))
+	}
+	if len(m.Partitions) > 0 {
+		l = 0
+		for _, e := range m.Partitions {
+			l += sovInternal(uint64(e))
+		}
+		n += 1 + sovInternal(uint64(l)) + l
+	}
+	if m.Readonly {
+		n += 2
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func (m *NullableInt64) Size() (n int) {
 	if m == nil {
 		return 0
@@ -4427,6 +4631,10 @@ func (m *PropagatedRequest) Size() (n int) {
 		l = m.ResumeStreamOp.Size()
 		n += 1 + l + sovInternal(uint64(l))
 	}
+	if m.SetStreamReadonlyOp != nil {
+		l = m.SetStreamReadonlyOp.Size()
+		n += 1 + l + sovInternal(uint64(l))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -4994,6 +5202,42 @@ func (m *RaftLog) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 10:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field SetStreamReadonlyOp", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowInternal
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthInternal
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthInternal
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.SetStreamReadonlyOp == nil {
+				m.SetStreamReadonlyOp = &SetStreamReadonlyOp{}
+			}
+			if err := m.SetStreamReadonlyOp.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipInternal(dAtA[iNdEx:])
@@ -6313,6 +6557,188 @@ func (m *PublishActivityOp) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *SetStreamReadonlyOp) 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: SetStreamReadonlyOp: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: SetStreamReadonlyOp: 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 < 0 {
+				return ErrInvalidLengthInternal
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Stream = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 2:
+			if wireType == 0 {
+				var v int32
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowInternal
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					v |= int32(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				m.Partitions = append(m.Partitions, v)
+			} else if wireType == 2 {
+				var packedLen int
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowInternal
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					packedLen |= int(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				if packedLen < 0 {
+					return ErrInvalidLengthInternal
+				}
+				postIndex := iNdEx + packedLen
+				if postIndex < 0 {
+					return ErrInvalidLengthInternal
+				}
+				if postIndex > l {
+					return io.ErrUnexpectedEOF
+				}
+				var elementCount int
+				var count int
+				for _, integer := range dAtA[iNdEx:postIndex] {
+					if integer < 128 {
+						count++
+					}
+				}
+				elementCount = count
+				if elementCount != 0 && len(m.Partitions) == 0 {
+					m.Partitions = make([]int32, 0, elementCount)
+				}
+				for iNdEx < postIndex {
+					var v int32
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowInternal
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						v |= int32(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					m.Partitions = append(m.Partitions, v)
+				}
+			} else {
+				return fmt.Errorf("proto: wrong wireType = %d for field Partitions", wireType)
+			}
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Readonly", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowInternal
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.Readonly = bool(v != 0)
+		default:
+			iNdEx = preIndex
+			skippy, err := skipInternal(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if skippy < 0 {
+				return ErrInvalidLengthInternal
+			}
+			if (iNdEx + skippy) < 0 {
+				return ErrInvalidLengthInternal
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *NullableInt64) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
@@ -8339,6 +8765,42 @@ func (m *PropagatedRequest) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 9:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field SetStreamReadonlyOp", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowInternal
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthInternal
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthInternal
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.SetStreamReadonlyOp == nil {
+				m.SetStreamReadonlyOp = &SetStreamReadonlyOp{}
+			}
+			if err := m.SetStreamReadonlyOp.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipInternal(dAtA[iNdEx:])
diff --git a/server/protocol/internal.proto b/server/protocol/internal.proto
index fa2adce..35557ac 100644
--- a/server/protocol/internal.proto
+++ b/server/protocol/internal.proto
@@ -7,27 +7,29 @@ message ServerState {
 }
 
 enum Op {
-    CREATE_STREAM    = 0;
-    SHRINK_ISR       = 1;
-    REPORT_LEADER    = 2;
-    CHANGE_LEADER    = 3;
-    EXPAND_ISR       = 4;
-    DELETE_STREAM    = 5;
-    PAUSE_STREAM     = 6;
-    RESUME_STREAM    = 7;
-    PUBLISH_ACTIVITY = 8;
+    CREATE_STREAM       = 0;
+    SHRINK_ISR          = 1;
+    REPORT_LEADER       = 2;
+    CHANGE_LEADER       = 3;
+    EXPAND_ISR          = 4;
+    DELETE_STREAM       = 5;
+    PAUSE_STREAM        = 6;
+    RESUME_STREAM       = 7;
+    PUBLISH_ACTIVITY    = 8;
+    SET_STREAM_READONLY = 9;
 }
 
 message RaftLog {
-    Op                op                = 1;
-    CreateStreamOp    createStreamOp    = 2;
-    ShrinkISROp       shrinkISROp       = 3;
-    ChangeLeaderOp    changeLeaderOp    = 4;
-    ExpandISROp       expandISROp       = 5;
-    DeleteStreamOp    deleteStreamOp    = 6;
-    PauseStreamOp     pauseStreamOp     = 7;
-    ResumeStreamOp    resumeStreamOp    = 8;
-    PublishActivityOp publishActivityOp = 9;
+    Op                  op                  = 1;
+    CreateStreamOp      createStreamOp      = 2;
+    ShrinkISROp         shrinkISROp         = 3;
+    ChangeLeaderOp      changeLeaderOp      = 4;
+    ExpandISROp         expandISROp         = 5;
+    DeleteStreamOp      deleteStreamOp      = 6;
+    PauseStreamOp       pauseStreamOp       = 7;
+    ResumeStreamOp      resumeStreamOp      = 8;
+    PublishActivityOp   publishActivityOp   = 9;
+    SetStreamReadonlyOp setStreamReadonlyOp = 10;
 }
 
 message CreateStreamOp {
@@ -83,6 +85,12 @@ message PublishActivityOp {
     uint64 raftIndex = 1;
 }
 
+message SetStreamReadonlyOp {
+    string         stream     = 1;
+    repeated int32 partitions = 2;
+    bool           readonly   = 3;
+}
+
 message NullableInt64 {
     int64 value = 1; 
 }
@@ -159,14 +167,15 @@ message LeaderEpochOffsetResponse {
 }
 
 message PropagatedRequest {
-    Op             op             = 1;
-    CreateStreamOp createStreamOp = 2;
-    ShrinkISROp    shrinkISROp    = 3;
-    ReportLeaderOp reportLeaderOp = 4;
-    ExpandISROp    expandISROp    = 5;
-    DeleteStreamOp deleteStreamOp = 6;
-    PauseStreamOp  pauseStreamOp  = 7;
-    ResumeStreamOp resumeStreamOp = 8;
+    Op                  op                  = 1;
+    CreateStreamOp      createStreamOp      = 2;
+    ShrinkISROp         shrinkISROp         = 3;
+    ReportLeaderOp      reportLeaderOp      = 4;
+    ExpandISROp         expandISROp         = 5;
+    DeleteStreamOp      deleteStreamOp      = 6;
+    PauseStreamOp       pauseStreamOp       = 7;
+    ResumeStreamOp      resumeStreamOp      = 8;
+    SetStreamReadonlyOp setStreamReadonlyOp = 9;
 }
 
 message Error {
@@ -183,7 +192,8 @@ message PropagatedResponse {
     // Reserving = 6 for expandISRResp if needed.
     // Reserving = 7 for deleteStreamResp if needed.
     // Reserving = 8 for pauseStreamResp if needed.
-    // Reserving = 9 for resumePartitionResp if needed.
+    // Reserving = 9 for resumeStreamResp if needed.
+    // Reserving = 10 for setStreamReadonlyResp if needed.
 }
 
 message ServerInfoRequest {
diff --git a/server/server.go b/server/server.go
index 47d4f3b..56389df 100644
--- a/server/server.go
+++ b/server/server.go
@@ -580,6 +580,8 @@ func (s *Server) handlePropagatedRequest(m *nats.Msg) {
 		resp = s.handlePauseStream(req)
 	case proto.Op_RESUME_STREAM:
 		resp = s.handleResumeStream(req)
+	case proto.Op_SET_STREAM_READONLY:
+		resp = s.handleSetStreamReadonly(req)
 	default:
 		s.logger.Warnf("Unknown propagated request operation: %s", req.Op)
 		return
@@ -663,6 +665,16 @@ func (s *Server) handleResumeStream(req *proto.PropagatedRequest) *proto.Propaga
 	return resp
 }
 
+func (s *Server) handleSetStreamReadonly(req *proto.PropagatedRequest) *proto.PropagatedResponse {
+	resp := &proto.PropagatedResponse{
+		Op: req.Op,
+	}
+	if err := s.metadata.SetStreamReadonly(context.Background(), req.SetStreamReadonlyOp); err != nil {
+		resp.Error = &proto.Error{Code: uint32(err.Code()), Msg: err.Message()}
+	}
+	return resp
+}
+
 func (s *Server) isShutdown() bool {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
diff --git a/server/stream.go b/server/stream.go
index 83c3386..a280855 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -151,3 +151,31 @@ func (s *stream) Delete() error {
 	}
 	return nil
 }
+
+// SetReadonly sets the readonly flag on some or all the partitions of this
+// stream.
+func (s *stream) SetReadonly(partitions []int32, readonly bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	toSetReadonly := make([]*partition, 0, len(partitions))
+	if len(partitions) == 0 {
+		for _, partition := range s.partitions {
+			toSetReadonly = append(toSetReadonly, partition)
+		}
+	} else {
+		for _, partitionID := range partitions {
+			partition, ok := s.partitions[partitionID]
+			if !ok {
+				return ErrPartitionNotFound
+			}
+			toSetReadonly = append(toSetReadonly, partition)
+		}
+	}
+
+	for _, partition := range toSetReadonly {
+		partition.SetReadonly(readonly)
+	}
+
+	return nil
+}


Test patch:
diff --git a/server/activity_test.go b/server/activity_test.go
index 2b3793a..c8742e5 100644
--- a/server/activity_test.go
+++ b/server/activity_test.go
@@ -226,6 +226,119 @@ func TestActivityStreamResumeStream(t *testing.T) {
 	}
 }
 
+// Ensure activity stream set readonly (on) event occurs.
+func TestActivityStreamSetStreamReadonly(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure server.
+	s1Config := getTestConfig("a", true, 5050)
+	s1Config.ActivityStream.Enabled = true
+	s1Config.ActivityStream.PublishTimeout = time.Second
+	s1Config.ActivityStream.PublishAckPolicy = liftApi.AckPolicy_LEADER
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	// Wait for server to elect itself leader.
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	// Create stream.
+	stream := "foo-stream"
+	require.NoError(t, client.CreateStream(context.Background(), "foo", stream))
+
+	// Set stream readonly.
+	require.NoError(t, client.SetStreamReadonly(context.Background(), stream))
+
+	// The first message read back should be the stream readonly.
+	msgs := make(chan *lift.Message, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	err = client.Subscribe(ctx, activityStream, func(msg *lift.Message, err error) {
+		require.NoError(t, err)
+		msgs <- msg
+		cancel()
+	}, lift.StartAtLatestReceived())
+	require.NoError(t, err)
+
+	// Wait to get the message.
+	select {
+	case msg := <-msgs:
+		var se liftApi.ActivityStreamEvent
+		err = se.Unmarshal(msg.Value())
+		require.NoError(t, err)
+		require.Equal(t, liftApi.ActivityStreamOp_SET_STREAM_READONLY, se.GetOp())
+		require.Equal(t, stream, se.SetStreamReadonlyOp.GetStream())
+		require.Equal(t, []int32{0}, se.SetStreamReadonlyOp.GetPartitions())
+		require.Equal(t, true, se.SetStreamReadonlyOp.GetReadonly())
+	case <-time.After(5 * time.Second):
+		t.Fatal("Did not receive expected message")
+	}
+}
+
+// Ensure activity stream set readonly (off) event occurs.
+func TestActivityStreamSetStreamReadwrite(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure server.
+	s1Config := getTestConfig("a", true, 5050)
+	s1Config.ActivityStream.Enabled = true
+	s1Config.ActivityStream.PublishTimeout = time.Second
+	s1Config.ActivityStream.PublishAckPolicy = liftApi.AckPolicy_LEADER
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	// Wait for server to elect itself leader.
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	// Create stream.
+	stream := "foo-stream"
+	require.NoError(t, client.CreateStream(context.Background(), "foo", stream))
+
+	// Set stream readonly.
+	require.NoError(t, client.SetStreamReadonly(context.Background(), stream))
+
+	// Set stream readwrite.
+	require.NoError(t, client.SetStreamReadonly(context.Background(), stream, lift.Readonly(false)))
+
+	// The first message read back should be the stream readwrite.
+	msgs := make(chan *lift.Message, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	err = client.Subscribe(ctx, activityStream, func(msg *lift.Message, err error) {
+		require.NoError(t, err)
+		msgs <- msg
+		cancel()
+	}, lift.StartAtLatestReceived())
+	require.NoError(t, err)
+
+	// Wait to get the message.
+	select {
+	case msg := <-msgs:
+		var se liftApi.ActivityStreamEvent
+		err = se.Unmarshal(msg.Value())
+		require.NoError(t, err)
+		require.Equal(t, liftApi.ActivityStreamOp_SET_STREAM_READONLY, se.GetOp())
+		require.Equal(t, stream, se.SetStreamReadonlyOp.GetStream())
+		require.Equal(t, []int32{0}, se.SetStreamReadonlyOp.GetPartitions())
+		require.Equal(t, false, se.SetStreamReadonlyOp.GetReadonly())
+	case <-time.After(5 * time.Second):
+		t.Fatal("Did not receive expected message")
+	}
+}
+
 // Ensure computeActivityPublishBackoff doubles the backoff time and caps it at
 // the max backoff.
 func TestComputeActivityPublishBackoff(t *testing.T) {
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index e771316..390f107 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -588,6 +588,113 @@ func TestNotifyLEOIdempotent(t *testing.T) {
 	require.Equal(t, ch1, ch2)
 }
 
+// Ensure when SetReadonly is called with true on a log, Append returns
+// ErrCommitLogReadonly. When SetReadonly is called with false, Appends
+// succeed.
+func TestSetReadonlyAppend(t *testing.T) {
+	l, cleanup := setupWithOptions(t, Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 256,
+	})
+	defer l.Close()
+	defer cleanup()
+
+	require.False(t, l.IsReadonly())
+
+	l.SetReadonly(true)
+	require.True(t, l.IsReadonly())
+
+	_, err := l.Append(msgs)
+	require.Equal(t, ErrCommitLogReadonly, err)
+
+	l.SetReadonly(false)
+	require.False(t, l.IsReadonly())
+
+	_, err = l.Append(msgs)
+	require.NoError(t, err)
+}
+
+// Ensure when SetReadonly is called with true on a log, committed readers
+// receive ErrCommitLogReadonly once they reach the LEO.
+func TestSetReadonlyReadToLEO(t *testing.T) {
+	l, cleanup := setupWithOptions(t, Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 256,
+	})
+	defer l.Close()
+	defer cleanup()
+
+	_, err := l.Append(msgs)
+	require.NoError(t, err)
+	r, err := l.NewReader(0, false)
+	require.NoError(t, err)
+	l.SetHighWatermark(4)
+
+	l.SetReadonly(true)
+
+	headers := make([]byte, 28)
+	for range msgs {
+		_, _, _, _, err := r.ReadMessage(context.Background(), headers)
+		require.NoError(t, err)
+	}
+
+	_, _, _, _, err = r.ReadMessage(context.Background(), headers)
+	require.Equal(t, ErrCommitLogReadonly, err)
+}
+
+// Ensure when SetReadonly is called with true on a log, committed readers
+// waiting for the HW to advance do not receive ErrCommitLogReadonly when the
+// HW is less than the LEO.
+func TestSetReadonlyDoNotWakeHWLessThanLEO(t *testing.T) {
+	l, cleanup := setupWithOptions(t, Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 256,
+	})
+	defer l.Close()
+	defer cleanup()
+
+	_, err := l.Append(msgs)
+	require.NoError(t, err)
+	r, err := l.NewReader(0, false)
+	require.NoError(t, err)
+
+	go func() {
+		time.Sleep(5 * time.Millisecond)
+		l.SetReadonly(true)
+		l.SetHighWatermark(4)
+	}()
+
+	headers := make([]byte, 28)
+	for range msgs {
+		_, _, _, _, err := r.ReadMessage(context.Background(), headers)
+		require.NoError(t, err)
+	}
+}
+
+// Ensure when SetReadonly is called with true on a log, committed readers
+// waiting for the HW to advance receive ErrCommitLogReadonly when the HW is
+// caught up to the LEO.
+func TestSetReadonlyWakeHWEqualsLEO(t *testing.T) {
+	l, cleanup := setupWithOptions(t, Options{
+		Path:            tempDir(t),
+		MaxSegmentBytes: 256,
+	})
+	defer l.Close()
+	defer cleanup()
+
+	r, err := l.NewReader(0, false)
+	require.NoError(t, err)
+
+	go func() {
+		time.Sleep(5 * time.Millisecond)
+		l.SetReadonly(true)
+	}()
+
+	headers := make([]byte, 28)
+	_, _, _, _, err = r.ReadMessage(context.Background(), headers)
+	require.Equal(t, ErrCommitLogReadonly, err)
+}
+
 func setup(t require.TestingT) (*commitLog, func()) {
 	opts := Options{
 		Path:            tempDir(t),
diff --git a/server/metadata_test.go b/server/metadata_test.go
index 8514268..cd3f2bd 100644
--- a/server/metadata_test.go
+++ b/server/metadata_test.go
@@ -256,3 +256,32 @@ func TestEnsureTimeout(t *testing.T) {
 	require.True(t, ok)
 	require.InDelta(t, timeout.Milliseconds(), time.Until(deadline).Milliseconds(), 1)
 }
+
+// Ensure checkResumeStreamPreconditions returns ErrPartitionNotFound if the
+// partition doesn't exist.
+func TestMetadataCheckSetStreamReadonlyPreconditionsPartitionNotFound(t *testing.T) {
+	defer cleanupStorage(t)
+
+	server := New(getTestConfig("a", true, 0))
+	metadata := newMetadataAPI(server)
+	defer metadata.Reset()
+
+	_, err := metadata.AddStream(&proto.Stream{
+		Name:    "foo",
+		Subject: "foo",
+		Partitions: []*proto.Partition{
+			{
+				Stream:  "foo",
+				Subject: "foo",
+				Id:      0,
+			},
+		},
+	}, false)
+	require.NoError(t, err)
+
+	err = metadata.checkSetStreamReadonlyPreconditions(&proto.RaftLog{
+		Op:                  proto.Op_SET_STREAM_READONLY,
+		SetStreamReadonlyOp: &proto.SetStreamReadonlyOp{Stream: "foo", Partitions: []int32{1}},
+	})
+	require.Equal(t, ErrPartitionNotFound, err)
+}
diff --git a/server/server_test.go b/server/server_test.go
index f517981..57f0ee3 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -136,6 +136,26 @@ func checkPartitionPaused(t *testing.T, timeout time.Duration, stream string,
 	stackFatalf(t, "Expected partition paused %v, got %v", paused, isPaused)
 }
 
+func checkPartitionReadonly(t *testing.T, timeout time.Duration, stream string,
+	partitionID int32, readonly bool, server *Server) {
+
+	partition := server.metadata.GetPartition(stream, partitionID)
+	if partition == nil {
+		stackFatalf(t, "Partition not found")
+	}
+	var (
+		deadline   = time.Now().Add(timeout)
+		isReadonly bool
+	)
+	for time.Now().Before(deadline) {
+		isReadonly = partition.IsReadonly()
+		if isReadonly == readonly {
+			return
+		}
+	}
+	stackFatalf(t, "Expected partition readonly %v, got %v", readonly, isReadonly)
+}
+
 func getPartitionLeader(t *testing.T, timeout time.Duration, name string, partitionID int32, servers ...*Server) *Server {
 	var (
 		leader   *Server
@@ -1549,86 +1569,141 @@ func TestPauseStreamPropagate(t *testing.T) {
 	checkPartitionPaused(t, 5*time.Second, name, 0, false, s2)
 }
 
-// Ensure the ISR is maintained across partition pauses.
-func TestPauseStreamMaintainISR(t *testing.T) {
+// Test setting and removing the readonly flag on a stream. This test sets all
+// partitions on a stream as readonly and checks that both cannot be published
+// to.
+func TestSetStreamReadonlyAllPartitions(t *testing.T) {
 	defer cleanupStorage(t)
 
 	// Use a central NATS server.
 	ns := natsdTest.RunDefaultServer()
 	defer ns.Shutdown()
 
-	// Configure first server.
+	// Configure server.
 	s1Config := getTestConfig("a", true, 5050)
-	s1Config.Clustering.ReplicaMaxLagTime = 2 * time.Second
-	s1Config.Clustering.ReplicaMaxIdleWait = 0
-	s1Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
 	s1 := runServerWithConfig(t, s1Config)
 	defer s1.Stop()
 
-	// Configure second server.
-	s2Config := getTestConfig("b", false, 0)
-	s2Config.Clustering.ReplicaMaxLagTime = 2 * time.Second
-	s2Config.Clustering.ReplicaMaxIdleWait = 0
-	s2Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
-	s2 := runServerWithConfig(t, s2Config)
-	defer s2.Stop()
+	// Wait for server to elect itself leader.
+	getMetadataLeader(t, 10*time.Second, s1)
 
-	// Configure third server.
-	s3Config := getTestConfig("c", false, 0)
-	s3Config.Clustering.ReplicaMaxLagTime = 2 * time.Second
-	s3Config.Clustering.ReplicaMaxIdleWait = 0
-	s3Config.Clustering.ReplicaMaxLeaderTimeout = time.Second
-	s3 := runServerWithConfig(t, s3Config)
-	defer s3.Stop()
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	// Create stream.
+	name := "foo"
+	subject := "foo"
+	err = client.CreateStream(context.Background(), subject, name, lift.Partitions(2))
+	require.NoError(t, err)
+
+	// Set the stream as readonly.
+	err = client.SetStreamReadonly(context.Background(), name)
+	require.NoError(t, err)
+
+	// Check that both partitions are readonly.
+	checkPartitionReadonly(t, 5*time.Second, name, 0, true, s1)
+	checkPartitionReadonly(t, 5*time.Second, name, 1, true, s1)
+
+	// Check that both partitions cannot be published to.
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err = client.Publish(ctx, name, []byte("hello"), lift.ToPartition(0))
+	require.Error(t, err)
+
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err = client.Publish(ctx, name, []byte("hello"), lift.ToPartition(1))
+	require.Error(t, err)
+}
+
+// Test setting and removing the readonly flag on a partition. A readonly
+// partition should not accept new publications but should still be
+// subscribable. This test sets only one partition as readonly and checks that
+// the other one can still be published to.
+func TestSetStreamReadonlySomePartitions(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure server.
+	s1Config := getTestConfig("a", true, 5050)
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
 
 	// Wait for server to elect itself leader.
-	getMetadataLeader(t, 10*time.Second, s1, s2, s3)
+	getMetadataLeader(t, 10*time.Second, s1)
 
 	client, err := lift.Connect([]string{"localhost:5050"})
 	require.NoError(t, err)
 	defer client.Close()
 
-	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-	defer cancel()
+	// Create stream.
 	name := "foo"
-	err = client.CreateStream(ctx, "foo", name, lift.ReplicationFactor(3))
+	subject := "foo"
+	err = client.CreateStream(context.Background(), subject, name, lift.Partitions(2))
 	require.NoError(t, err)
 
-	// Check the ISR size is 3.
-	waitForISR(t, 5*time.Second, name, 0, 3, s1, s2, s3)
+	// Try to set a non-existing stream as readonly.
+	err = client.SetStreamReadonly(context.Background(), "bar")
+	require.Error(t, err)
 
-	// Kill a replica.
-	s2.Stop()
+	// Try to set a non-existing partition as readonly.
+	err = client.SetStreamReadonly(context.Background(), name, lift.ReadonlyPartitions(99))
+	require.Error(t, err)
 
-	// Ensure ISR size shrinks to 2.
-	waitForISR(t, 10*time.Second, name, 0, 2, s1, s3)
+	// Set one partition as readonly.
+	err = client.SetStreamReadonly(context.Background(), name, lift.ReadonlyPartitions(0))
+	require.NoError(t, err)
 
-	// Pause stream.
-	err = client.PauseStream(context.Background(), name)
+	// Make sure setting readonly is idempotent.
+	err = client.SetStreamReadonly(context.Background(), name, lift.ReadonlyPartitions(0))
 	require.NoError(t, err)
 
-	checkPartitionPaused(t, 5*time.Second, name, 0, true, s1)
-	checkPartitionPaused(t, 5*time.Second, name, 0, true, s3)
+	// Check that only the first partition is readonly.
+	checkPartitionReadonly(t, 5*time.Second, name, 0, true, s1)
+	checkPartitionReadonly(t, 5*time.Second, name, 1, false, s1)
 
-	// Resume stream by publishing.
-	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)
+	// Publish a message to partition 0.
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	_, err = client.Publish(ctx, name, []byte("hello"))
+	require.Error(t, err)
+
+	// Publish a message to partition 1.
+	ctx, cancel = context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	_, err = client.Publish(ctx, name, []byte("hello"), lift.ToPartition(1))
 	require.NoError(t, err)
 
-	// Ensure stream is resumed.
-	checkPartitionPaused(t, 5*time.Second, name, 0, false, s1)
-	checkPartitionPaused(t, 5*time.Second, name, 0, false, s3)
+	msgs := make(chan *lift.Message, 1)
+	ctx, cancel = context.WithCancel(context.Background())
+	err = client.Subscribe(ctx, name, func(msg *lift.Message, err error) {
+		require.NoError(t, err)
+		msgs <- msg
+		cancel()
+	}, lift.StartAtEarliestReceived(), lift.Partition(1))
+	require.NoError(t, err)
+
+	// Wait to get the new message.
+	select {
+	case msg := <-msgs:
+		require.Equal(t, []byte("hello"), msg.Value())
+	case <-time.After(5 * time.Second):
+		t.Fatal("Did not receive expected message")
+	}
 
-	// Ensure ISR size is still 2.
-	waitForISR(t, 10*time.Millisecond, name, 0, 2, s1, s3)
+	// Check that only the first partition is readonly.
+	checkPartitionReadonly(t, 5*time.Second, name, 0, true, s1)
+	checkPartitionReadonly(t, 5*time.Second, name, 1, false, s1)
 }
 
-// Test stream pausing and resuming by subscription. A paused stream should
-// re-activate itself when it is subscribed to (with the Resume option
-// enabled) using the Liftbridge API. This test pauses all partitions and checks
-// that only the partition that is subscribed to gets resumed.
-func TestPauseStreamResumeBySubscription(t *testing.T) {
+// Test subscribing to a readonly stream. Stream messages should be received,
+// followed by an ErrEndOfReadonlyPartition error that also ends the
+// subscription.
+func TestSetStreamReadonlySubscription(t *testing.T) {
 	defer cleanupStorage(t)
 
 	// Use a central NATS server.
@@ -1653,37 +1728,83 @@ func TestPauseStreamResumeBySubscription(t *testing.T) {
 	err = client.CreateStream(context.Background(), subject, name, lift.Partitions(2))
 	require.NoError(t, err)
 
-	// Publish a message to the first partition.
+	// Publish a message to the stream.
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
-	_, err = client.Publish(ctx, name, []byte("hello"), lift.ToPartition(0))
+	_, err = client.Publish(ctx, name, []byte("hello"))
 	require.NoError(t, err)
 
-	// Pause all partitions.
-	err = client.PauseStream(context.Background(), name)
+	// Set stream as readonly.
+	err = client.SetStreamReadonly(context.Background(), name)
 	require.NoError(t, err)
 
-	// Subscribe to partition 0.
 	msgs := make(chan *lift.Message, 1)
 	ctx, cancel = context.WithCancel(context.Background())
 	err = client.Subscribe(ctx, name, func(msg *lift.Message, err error) {
-		require.NoError(t, err)
+		// Ignore published messages, we are only waiting for the end of
+		// readonly partition error.
+		if err == nil {
+			return
+		}
+
+		require.EqualError(t, err, lift.ErrReadonlyPartition.Error())
 		msgs <- msg
 		cancel()
-	}, lift.StartAtEarliestReceived(), lift.Resume(), lift.Partition(0))
+	}, lift.StartAtEarliestReceived())
 	require.NoError(t, err)
 
-	// Wait to get the new message.
+	// Wait to get the error.
 	select {
-	case msg := <-msgs:
-		require.Equal(t, []byte("hello"), msg.Value())
+	case <-msgs:
 	case <-time.After(5 * time.Second):
-		t.Fatal("Did not receive expected message")
+		t.Fatal("Did not receive expected error")
 	}
+}
 
-	// Check that partition 0 was resumed but partition 1 is still paused.
-	checkPartitionPaused(t, 5*time.Second, name, 0, false, s1)
-	checkPartitionPaused(t, 5*time.Second, name, 1, true, s1)
+// Ensure setting a stream readonly works when we send the request to the
+// metadata follower and removing the readonly flag on the stream works when
+// the SetReadonly call is sent to the follower.
+func TestSetStreamReadonlyPropagate(t *testing.T) {
+	defer cleanupStorage(t)
+
+	// Use a central NATS server.
+	ns := natsdTest.RunDefaultServer()
+	defer ns.Shutdown()
+
+	// Configure first server.
+	s1Config := getTestConfig("a", true, 0)
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	// Configure second server.
+	s2Config := getTestConfig("b", false, 5050)
+	s2 := runServerWithConfig(t, s2Config)
+	defer s2.Stop()
+
+	// Connect and send the request to the follower.
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	name := "foo"
+	err = client.CreateStream(ctx, "foo", name)
+	require.NoError(t, err)
+
+	// Set the stream as readonly on follower.
+	err = client.SetStreamReadonly(context.Background(), name)
+	require.NoError(t, err)
+
+	checkPartitionReadonly(t, 5*time.Second, name, 0, true, s1)
+	checkPartitionReadonly(t, 5*time.Second, name, 0, true, s2)
+
+	// Remove the readonly flag from the stream on follower.
+	err = client.SetStreamReadonly(context.Background(), name, lift.Readonly(false))
+	require.NoError(t, err)
+
+	checkPartitionReadonly(t, 5*time.Second, name, 0, false, s1)
+	checkPartitionReadonly(t, 5*time.Second, name, 0, false, s2)
 }
 
 // Ensure publishing to a non-existent stream returns an error.