Repository: liftbridge-io/liftbridge
Pull request #296 (2020-12-13 12:50:38)

Add support for Optimistic Concurrency Control
Depends on : https://github.com/liftbridge-io/go-liftbridge/pull/101 and https://github.com/liftbridge-io/liftbridge-api/pull/48

To support [Optimistic Concurrency Control](https://github.com/liftbridge-io/liftbridge/issues/54), the following changes are needed:

- [x]  Add config optin for `OptimisticConcurrencyControl` on stream level. This config is passed down to `commitLog` level
- [x]  In `dispatchAcks` check for `AckError ` and do a `sendPublishAsyncError`
- [x]  The actual concurrency control is performed on on `message_set.newMessageSetFromProto()`. This is called in `messageProcessingLoop`. 
- [x]  Add tests 
- [x]  Update docs
- [x]    Should `sendPublishAsyncError` returns predefined types of error for each type of `AckError` available ?

The overall operations of concurrency control is resumed in the following schema:

![ooc](https://user-images.githubusercontent.com/6369285/102012280-c6136980-3d49-11eb-9ea8-13aa67953d75.png)





Patch:
diff --git a/documentation/client_implementation.md b/documentation/client_implementation.md
index 2272217..8679c96 100644
--- a/documentation/client_implementation.md
+++ b/documentation/client_implementation.md
@@ -1210,6 +1210,7 @@ SegmentMaxBytes
 SegmentMaxAge
 CompactEnabled
 CompactMaxGoroutines
+OptimisticConcurrencyControl
 ```
 
 Refer to [Stream Configuration](configuration.md#streams-configuration-settings)
@@ -1241,6 +1242,11 @@ explicitly, the servier will use the default configuration and that may be to
 disable compaction on the service side, which renders `CompactMaxGoroutines` to
 be unused.
 
+Note: if `OptimisticConcurrencyControl` is configured, you should to make sure
+that at least one `AckPolicy` is set. A `PublishRequest` with `AckPolicy` is set to `AckPolicy_NONE`
+will be rejected explicity by the server.
+
+
 ```go
 // CreateStream creates a new stream attached to a NATS subject. Subject is the
 // NATS subject the stream is attached to, and name is the stream identifier,
diff --git a/documentation/concepts.md b/documentation/concepts.md
index a22b8dc..7b24d05 100644
--- a/documentation/concepts.md
+++ b/documentation/concepts.md
@@ -384,6 +384,18 @@ serializing their messages into [*envelopes*](./envelope_protocol.md). An
 envelope allows publishers to set things like the `AckInbox`, `Key`, `Headers`,
 and other pieces of metadata.
 
+## Concurrency Control
+Streams support control concurrency publishes. This is achieved by sending `ExpectedOffset`
+on every publish request. The server may check and only approve message that has correct
+`ExpectedOffset`. It faciliates the control when multiple clients publish to the same stream.
+
+This feature enables complex use cases such as:
+
+- Idempotent publish
+- Use the stream as commitlog
+
+This behavior may be enabled by [*configuration*](./configuration.md)
+
 ## Further Reading
 
 A final note is to read ["Designing Event-Driven
diff --git a/documentation/configuration.md b/documentation/configuration.md
index b7d2c30..3eafe80 100644
--- a/documentation/configuration.md
+++ b/documentation/configuration.md
@@ -181,6 +181,7 @@ overriding these settings.
 | compact.max.goroutines | | The maximum number of concurrent goroutines to use for compaction on a stream log (only applicable if `compact.enabled` is `true`). | int | 10 | |
 | auto.pause.time | | The amount of time a stream partition can go idle, i.e. not receive a message, before it is automatically paused. A value of 0 disables auto pausing. | duration | 0 | |
 | auto.pause.disable.if.subscribers | | Disables automatic stream partition pausing when there are subscribers. | bool | false | |
+| concurrency.control | | Enable Optimistic Concurrency Control on message publishing for all streams. | bool | false | |
 
 ### Clustering Configuration Settings
 
diff --git a/go.mod b/go.mod
index 9b85328..2ce6318 100644
--- a/go.mod
+++ b/go.mod
@@ -15,7 +15,7 @@ require (
 	github.com/hashicorp/go-msgpack v1.1.5 // indirect
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/hashicorp/raft v1.1.2
-	github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201118225953-b849cccb6467
+	github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201230170826-7ed0a0b06362
 	github.com/liftbridge-io/liftbridge-api v1.4.2-0.20201228201911-4b2d99797dbb
 	github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3
 	github.com/liftbridge-io/raft-boltdb v0.0.0-20200414234651-aaf6e08d8f73
diff --git a/go.sum b/go.sum
index c6cc478..6c1abc0 100644
--- a/go.sum
+++ b/go.sum
@@ -190,8 +190,11 @@ 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.20201118225953-b849cccb6467 h1:Ls62H5vuS7lNfBdDnLaA0kG8hIa91fNJRQY2WGYV/aM=
 github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201118225953-b849cccb6467/go.mod h1:24NMu02Ba2sMO2y+IYstP1UFKVA4a6/p54Lc7KccSLc=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201230170826-7ed0a0b06362 h1:oRrmcbPJxSaS1YaGqLlSOUwTLn1/8shN/qeC3pSKCDA=
+github.com/liftbridge-io/go-liftbridge/v2 v2.0.2-0.20201230170826-7ed0a0b06362/go.mod h1:1yLlNSFpRxHAUiFlmBC57h1FDZbkx6ypoMximsIpIhU=
 github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201029165056-10f2aa65f256 h1:2pZtC3v6IBTwE70xfb/k0DPlOJ6BlXpthCUWrxCnhwo=
 github.com/liftbridge-io/liftbridge-api v1.1.1-0.20201029165056-10f2aa65f256/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
+github.com/liftbridge-io/liftbridge-api v1.3.1-0.20201124191421-8ceb6d0858ac/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
 github.com/liftbridge-io/liftbridge-api v1.4.2-0.20201228201911-4b2d99797dbb h1:XdwKrJh9gUJ+kaOwYfU6Kt+mZxqOeJQ8oLFbVjpoNuA=
 github.com/liftbridge-io/liftbridge-api v1.4.2-0.20201228201911-4b2d99797dbb/go.mod h1:6IFEFZ4ncnOgeDVjSt0vh1lKNhlJ5YT9xnG1eRa9LC8=
 github.com/liftbridge-io/nats-on-a-log v0.0.0-20200818183806-bb17516cf3a3 h1:O4mg1NEmukgY8hxen3grrG5RY34LadMTzpbjf8kM2tA=
diff --git a/server/api.go b/server/api.go
index 7d2535d..48ed5a1 100644
--- a/server/api.go
+++ b/server/api.go
@@ -282,8 +282,7 @@ func (a *apiServer) Publish(ctx context.Context, req *client.PublishRequest) (
 		return nil, convertPublishAsyncError(e)
 	}
 
-	if e := a.ensureStreamNotReadonly(req.Stream, req.Partition); e != nil {
-		a.logger.Errorf("api: Failed to publish message: %v", e.Message)
+	if e := a.ensurePublishPreconditions(req); e != nil {
 		return nil, convertPublishAsyncError(e)
 	}
 
@@ -306,6 +305,7 @@ func (a *apiServer) Publish(ctx context.Context, req *client.PublishRequest) (
 			AckInbox:      req.AckInbox,
 			CorrelationId: req.CorrelationId,
 			AckPolicy:     req.AckPolicy,
+			Offset:        req.ExpectedOffset,
 		}
 		resp = new(client.PublishResponse)
 	)
@@ -425,14 +425,21 @@ func (a *apiServer) FetchCursor(ctx context.Context, req *client.FetchCursorRequ
 	return &client.FetchCursorResponse{Offset: offset}, nil
 }
 
-func (a *apiServer) ensureStreamNotReadonly(name string, partitionID int32) *client.PublishAsyncError {
+func (a *apiServer) ensurePublishPreconditions(req *client.PublishRequest) *client.PublishAsyncError {
+	name := req.Stream
+	partitionID := req.Partition
+
 	stream := a.metadata.GetStream(name)
+
+	// Verify stream exists
 	if stream == nil {
 		return &client.PublishAsyncError{
 			Code:    client.PublishAsyncError_NOT_FOUND,
 			Message: fmt.Sprintf("no such stream: %s", name),
 		}
 	}
+
+	// Verify partition exists
 	partition := stream.GetPartition(partitionID)
 	if partition == nil {
 		return &client.PublishAsyncError{
@@ -440,6 +447,8 @@ func (a *apiServer) ensureStreamNotReadonly(name string, partitionID int32) *cli
 			Message: fmt.Sprintf("no such partition: %d", partitionID),
 		}
 	}
+
+	// Verify stream is not read only
 	if partition.IsReadonly() {
 		return &client.PublishAsyncError{
 			Code:    client.PublishAsyncError_READONLY,
@@ -447,6 +456,14 @@ func (a *apiServer) ensureStreamNotReadonly(name string, partitionID int32) *cli
 		}
 	}
 
+	// Verify AckPolicy is set for streams with Optimistic Concurrency Control
+	if partition.log.IsConcurrencyControlEnabled() && req.AckPolicy == client.AckPolicy_NONE {
+		return &client.PublishAsyncError{
+			Code:    client.PublishAsyncError_BAD_REQUEST,
+			Message: fmt.Sprintf("stream with concurrency control must have AckPolicy set"),
+		}
+	}
+
 	return nil
 }
 
@@ -784,6 +801,9 @@ func getStreamConfig(req *client.CreateStreamRequest) *proto.StreamConfig {
 	if req.MinIsr != nil {
 		config.MinIsr = &proto.NullableInt32{Value: req.MinIsr.Value}
 	}
+	if req.OptimisticConcurrencyControl != nil {
+		config.OptimisticConcurrencyControl = &proto.NullableBool{Value: req.OptimisticConcurrencyControl.Value}
+	}
 	return config
 }
 
@@ -899,7 +919,7 @@ func (p *publishAsyncSession) publishLoop() error {
 			return err
 		}
 
-		if e := p.ensureStreamNotReadonly(req.Stream, req.Partition); e != nil {
+		if e := p.ensurePublishPreconditions(req); e != nil {
 			p.logger.Errorf("api: Failed to publish async message: %v", e.Message)
 			p.sendPublishAsyncError(req.CorrelationId, e)
 			continue
@@ -933,6 +953,7 @@ func (p *publishAsyncSession) publishLoop() error {
 			AckInbox:      req.AckInbox,
 			CorrelationId: req.CorrelationId,
 			AckPolicy:     req.AckPolicy,
+			Offset:        req.ExpectedOffset,
 		})
 		if err != nil {
 			err = errors.Wrap(err, "failed to marshal message")
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index f6f1134..602c4e6 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -22,6 +22,10 @@ import (
 // ErrSegmentNotFound is returned if the segment could not be found.
 var ErrSegmentNotFound = errors.New("segment not found")
 
+// ErrIncorrectOffset is returned if the offset is incorrect. This is used in case Optimistic
+// Concurrency Control is activated.
+var ErrIncorrectOffset = errors.New("incorrect offset")
+
 const (
 	logFileSuffix               = ".log"
 	indexFileSuffix             = ".index"
@@ -62,6 +66,7 @@ type Options struct {
 	CompactMaxGoroutines int           // Max number of goroutines to use in a log compaction
 	CleanerInterval      time.Duration // Frequency to enforce retention policy
 	HWCheckpointInterval time.Duration // Frequency to checkpoint HW to disk
+	ConcurrencyControl   bool          // Optimistic Concurrency Control
 	Logger               logger.Logger
 }
 
@@ -225,7 +230,7 @@ func (l *commitLog) Append(msgs []*Message) ([]int64, error) {
 		segment          = l.activeSegment()
 		basePosition     = segment.Position()
 		baseOffset       = segment.NextOffset()
-		ms, entries, err = newMessageSetFromProto(baseOffset, basePosition, msgs)
+		ms, entries, err = newMessageSetFromProto(baseOffset, basePosition, msgs, l.IsConcurrencyControlEnabled())
 	)
 	if err != nil {
 		return nil, err
@@ -647,6 +652,11 @@ func (l *commitLog) IsReadonly() bool {
 	return atomic.LoadInt32(&l.readonly) == 1
 }
 
+// IsConcurrencyControlEnabled indicates if the log should check for concurrency before appending messages
+func (l *commitLog) IsConcurrencyControlEnabled() bool {
+	return l.Options.ConcurrencyControl
+}
+
 // 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 d75fcfa..7ac1b8a 100644
--- a/server/commitlog/interface.go
+++ b/server/commitlog/interface.go
@@ -89,6 +89,9 @@ type CommitLog interface {
 	// IsReadonly indicates if the log is in readonly mode.
 	IsReadonly() bool
 
+	// IsConcurrencyControlEnabled indicates if the log should check for concurrency before appending messages
+	IsConcurrencyControlEnabled() 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/message.go b/server/commitlog/message.go
index e74ee86..05e6257 100644
--- a/server/commitlog/message.go
+++ b/server/commitlog/message.go
@@ -24,6 +24,7 @@ type Message struct {
 	AckInbox      string
 	CorrelationID string
 	AckPolicy     client.AckPolicy
+	Offset        int64
 }
 
 // Encode the Message into the packetEncoder.
diff --git a/server/commitlog/message_set.go b/server/commitlog/message_set.go
index 5ac9359..3ec532f 100644
--- a/server/commitlog/message_set.go
+++ b/server/commitlog/message_set.go
@@ -48,9 +48,14 @@ func entriesForMessageSet(basePos int64, ms []byte) []*entry {
 	return entries
 }
 
-func newMessageSetFromProto(baseOffset, basePos int64, msgs []*Message) (
+func newMessageSetFromProto(baseOffset, basePos int64, msgs []*Message, concurrencyControl bool) (
 	messageSet, []*entry, error) {
 
+	// When concurrency control is enabled, messages shall be processed on by one
+	if concurrencyControl && len(msgs) > 1 {
+		panic(fmt.Errorf("Concurrency Control is enabled, unable to process a batch of messages"))
+	}
+
 	var (
 		buf     = new(bytes.Buffer)
 		entries = make([]*entry, len(msgs))
@@ -66,6 +71,14 @@ func newMessageSetFromProto(baseOffset, basePos int64, msgs []*Message) (
 			relPos = int64(n)
 			offset = int64(i) + baseOffset
 		)
+
+		// Check expected offset for concurrency in case of Optimistic Concurrency Control
+		if concurrencyControl && m.Offset != -1 {
+			if offset != m.Offset {
+				return nil, nil, ErrIncorrectOffset
+			}
+		}
+
 		if err := binary.Write(buf, encoding, uint64(offset)); err != nil {
 			return nil, nil, err
 		}
diff --git a/server/config.go b/server/config.go
index 53a1ae0..4167a96 100644
--- a/server/config.go
+++ b/server/config.go
@@ -52,6 +52,7 @@ const (
 	defaultActivityStreamPublishTimeout   = 5 * time.Second
 	defaultActivityStreamPublishAckPolicy = client.AckPolicy_ALL
 	defaultCursorsStreamAutoPauseTime     = time.Minute
+	defaultConcurrencyControl             = false
 )
 
 // Config setting key names.
@@ -94,6 +95,7 @@ const (
 	configStreamsCompactMaxGoroutines          = "streams.compact.max.goroutines"
 	configStreamsAutoPauseTime                 = "streams.auto.pause.time"
 	configStreamsAutoPauseDisableIfSubscribers = "streams.auto.pause.disable.if.subscribers"
+	configStreamsConcurrencyControl            = "streams.concurrency.control"
 
 	configClusteringServerID                = "clustering.server.id"
 	configClusteringNamespace               = "clustering.namespace"
@@ -149,6 +151,7 @@ var configKeys = map[string]struct{}{
 	configStreamsSegmentMaxBytes:               {},
 	configStreamsSegmentMaxAge:                 {},
 	configStreamsCompactEnabled:                {},
+	configStreamsConcurrencyControl:            {},
 	configStreamsCompactMaxGoroutines:          {},
 	configStreamsAutoPauseTime:                 {},
 	configStreamsAutoPauseDisableIfSubscribers: {},
@@ -186,6 +189,7 @@ type StreamsConfig struct {
 	AutoPauseTime                 time.Duration
 	AutoPauseDisableIfSubscribers bool
 	MinISR                        int
+	ConcurrencyControl            bool
 }
 
 // RetentionString returns a human-readable string representation of the
@@ -275,6 +279,10 @@ func (l *StreamsConfig) ApplyOverrides(c *proto.StreamConfig) {
 	if minISR := c.MinIsr; minISR != nil {
 		l.MinISR = int(minISR.Value)
 	}
+
+	if optimisticConcurrencyControl := c.OptimisticConcurrencyControl; optimisticConcurrencyControl != nil {
+		l.ConcurrencyControl = optimisticConcurrencyControl.Value
+	}
 }
 
 // ClusteringConfig contains settings for controlling cluster behavior.
@@ -361,6 +369,7 @@ func NewDefaultConfig() *Config {
 	config.Streams.SegmentMaxAge = defaultMaxSegmentAge
 	config.Streams.RetentionMaxAge = defaultRetentionMaxAge
 	config.Streams.CleanerInterval = defaultCleanerInterval
+	config.Streams.ConcurrencyControl = defaultConcurrencyControl
 	config.ActivityStream.PublishTimeout = defaultActivityStreamPublishTimeout
 	config.ActivityStream.PublishAckPolicy = defaultActivityStreamPublishAckPolicy
 	config.CursorsStream.AutoPauseTime = defaultCursorsStreamAutoPauseTime
@@ -668,7 +677,9 @@ func parseStreamsConfig(config *Config, v *viper.Viper) error {
 	if v.IsSet(configStreamsAutoPauseDisableIfSubscribers) {
 		config.Streams.AutoPauseDisableIfSubscribers = v.GetBool(configStreamsAutoPauseDisableIfSubscribers)
 	}
-
+	if v.IsSet(configStreamsConcurrencyControl) {
+		config.Streams.ConcurrencyControl = v.GetBool(configStreamsConcurrencyControl)
+	}
 	return nil
 }
 
diff --git a/server/partition.go b/server/partition.go
index 0e71bd1..5282eda 100644
--- a/server/partition.go
+++ b/server/partition.go
@@ -152,6 +152,7 @@ func (s *Server) newPartition(protoPartition *proto.Partition, recovered bool, c
 			Compact:              streamsConfig.Compact,
 			CompactMaxGoroutines: streamsConfig.CompactMaxGoroutines,
 			Logger:               s.logger,
+			ConcurrencyControl:   streamsConfig.ConcurrencyControl,
 		})
 	)
 	if err != nil {
@@ -778,6 +779,12 @@ func (p *partition) messageProcessingLoop(recvChan <-chan *nats.Msg, stop <-chan
 		batchWait = p.srv.config.BatchMaxTime
 		msgBatch  = make([]*commitlog.Message, 0, batchSize)
 	)
+	// If Concurrency Control is enabled, then the message will be appended one by one.
+	// This is to ensure no conflict between each message.
+	if p.log.IsConcurrencyControlEnabled() {
+		batchSize = 1
+	}
+
 	for {
 		msgBatch = msgBatch[:0]
 		select {
@@ -836,6 +843,23 @@ func (p *partition) messageProcessingLoop(recvChan <-chan *nats.Msg, stop <-chan
 		// Write uncommitted messages to log.
 		offsets, err := p.log.Append(msgBatch)
 		if err != nil {
+
+			// AckErr should be dispatched if ErrIncorrectOffset is raised.
+			if errors.Is(err, commitlog.ErrIncorrectOffset) {
+				msg := msgBatch[0]
+				ack := &client.Ack{
+					Stream:             p.Stream,
+					PartitionSubject:   p.Subject,
+					MsgSubject:         string(msg.Headers["subject"]),
+					AckInbox:           msg.AckInbox,
+					CorrelationId:      msg.CorrelationID,
+					AckPolicy:          msg.AckPolicy,
+					ReceptionTimestamp: msg.Timestamp,
+					AckError:           client.Ack_INCORRECT_OFFSET,
+				}
+
+				p.sendAck(ack)
+			}
 			p.srv.logger.Errorf("Failed to append to log %s: %v", p, err)
 			continue
 		}
@@ -1418,6 +1442,7 @@ func natsToProtoMessage(msg *nats.Msg, leaderEpoch uint64) *commitlog.Message {
 		m.AckInbox = message.AckInbox
 		m.CorrelationID = message.CorrelationId
 		m.AckPolicy = message.AckPolicy
+		m.Offset = message.Offset
 	} else {
 		m.Value = msg.Data
 	}
diff --git a/server/protocol/internal.pb.go b/server/protocol/internal.pb.go
index fd7a47a..37ff2e9 100644
--- a/server/protocol/internal.pb.go
+++ b/server/protocol/internal.pb.go
@@ -1,5 +1,5 @@
 // Code generated by protoc-gen-gogo. DO NOT EDIT.
-// source: server/protocol/internal.proto
+// source: internal.proto
 
 package protocol
 
@@ -20,7 +20,7 @@ var _ = math.Inf
 // is compatible with the proto package it is being compiled against.
 // A compilation error at this line likely means your copy of the
 // proto package needs to be updated.
-const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
 
 type Op int32
 
@@ -68,7 +68,7 @@ func (x Op) String() string {
 }
 
 func (Op) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{0}
+	return fileDescriptor_41f4a519b878ee3b, []int{0}
 }
 
 type ServerState struct {
@@ -82,7 +82,7 @@ func (m *ServerState) Reset()         { *m = ServerState{} }
 func (m *ServerState) String() string { return proto.CompactTextString(m) }
 func (*ServerState) ProtoMessage()    {}
 func (*ServerState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{0}
+	return fileDescriptor_41f4a519b878ee3b, []int{0}
 }
 func (m *ServerState) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -92,7 +92,7 @@ func (m *ServerState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
 		return xxx_messageInfo_ServerState.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -138,7 +138,7 @@ func (m *RaftLog) Reset()         { *m = RaftLog{} }
 func (m *RaftLog) String() string { return proto.CompactTextString(m) }
 func (*RaftLog) ProtoMessage()    {}
 func (*RaftLog) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{1}
+	return fileDescriptor_41f4a519b878ee3b, []int{1}
 }
 func (m *RaftLog) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -148,7 +148,7 @@ func (m *RaftLog) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 		return xxx_messageInfo_RaftLog.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -248,7 +248,7 @@ func (m *CreateStreamOp) Reset()         { *m = CreateStreamOp{} }
 func (m *CreateStreamOp) String() string { return proto.CompactTextString(m) }
 func (*CreateStreamOp) ProtoMessage()    {}
 func (*CreateStreamOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{2}
+	return fileDescriptor_41f4a519b878ee3b, []int{2}
 }
 func (m *CreateStreamOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -258,7 +258,7 @@ func (m *CreateStreamOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, erro
 		return xxx_messageInfo_CreateStreamOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -299,7 +299,7 @@ func (m *ShrinkISROp) Reset()         { *m = ShrinkISROp{} }
 func (m *ShrinkISROp) String() string { return proto.CompactTextString(m) }
 func (*ShrinkISROp) ProtoMessage()    {}
 func (*ShrinkISROp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{3}
+	return fileDescriptor_41f4a519b878ee3b, []int{3}
 }
 func (m *ShrinkISROp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -309,7 +309,7 @@ func (m *ShrinkISROp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
 		return xxx_messageInfo_ShrinkISROp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -378,7 +378,7 @@ func (m *ExpandISROp) Reset()         { *m = ExpandISROp{} }
 func (m *ExpandISROp) String() string { return proto.CompactTextString(m) }
 func (*ExpandISROp) ProtoMessage()    {}
 func (*ExpandISROp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{4}
+	return fileDescriptor_41f4a519b878ee3b, []int{4}
 }
 func (m *ExpandISROp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -388,7 +388,7 @@ func (m *ExpandISROp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
 		return xxx_messageInfo_ExpandISROp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -453,7 +453,7 @@ func (m *DeleteStreamOp) Reset()         { *m = DeleteStreamOp{} }
 func (m *DeleteStreamOp) String() string { return proto.CompactTextString(m) }
 func (*DeleteStreamOp) ProtoMessage()    {}
 func (*DeleteStreamOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{5}
+	return fileDescriptor_41f4a519b878ee3b, []int{5}
 }
 func (m *DeleteStreamOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -463,7 +463,7 @@ func (m *DeleteStreamOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, erro
 		return xxx_messageInfo_DeleteStreamOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -502,7 +502,7 @@ func (m *PauseStreamOp) Reset()         { *m = PauseStreamOp{} }
 func (m *PauseStreamOp) String() string { return proto.CompactTextString(m) }
 func (*PauseStreamOp) ProtoMessage()    {}
 func (*PauseStreamOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{6}
+	return fileDescriptor_41f4a519b878ee3b, []int{6}
 }
 func (m *PauseStreamOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -512,7 +512,7 @@ func (m *PauseStreamOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, error
 		return xxx_messageInfo_PauseStreamOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -564,7 +564,7 @@ func (m *ResumeStreamOp) Reset()         { *m = ResumeStreamOp{} }
 func (m *ResumeStreamOp) String() string { return proto.CompactTextString(m) }
 func (*ResumeStreamOp) ProtoMessage()    {}
 func (*ResumeStreamOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{7}
+	return fileDescriptor_41f4a519b878ee3b, []int{7}
 }
 func (m *ResumeStreamOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -574,7 +574,7 @@ func (m *ResumeStreamOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, erro
 		return xxx_messageInfo_ResumeStreamOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -622,7 +622,7 @@ func (m *ReportLeaderOp) Reset()         { *m = ReportLeaderOp{} }
 func (m *ReportLeaderOp) String() string { return proto.CompactTextString(m) }
 func (*ReportLeaderOp) ProtoMessage()    {}
 func (*ReportLeaderOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{8}
+	return fileDescriptor_41f4a519b878ee3b, []int{8}
 }
 func (m *ReportLeaderOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -632,7 +632,7 @@ func (m *ReportLeaderOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, erro
 		return xxx_messageInfo_ReportLeaderOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -699,7 +699,7 @@ func (m *ChangeLeaderOp) Reset()         { *m = ChangeLeaderOp{} }
 func (m *ChangeLeaderOp) String() string { return proto.CompactTextString(m) }
 func (*ChangeLeaderOp) ProtoMessage()    {}
 func (*ChangeLeaderOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{9}
+	return fileDescriptor_41f4a519b878ee3b, []int{9}
 }
 func (m *ChangeLeaderOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -709,7 +709,7 @@ func (m *ChangeLeaderOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, erro
 		return xxx_messageInfo_ChangeLeaderOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -760,7 +760,7 @@ func (m *PublishActivityOp) Reset()         { *m = PublishActivityOp{} }
 func (m *PublishActivityOp) String() string { return proto.CompactTextString(m) }
 func (*PublishActivityOp) ProtoMessage()    {}
 func (*PublishActivityOp) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{10}
+	return fileDescriptor_41f4a519b878ee3b, []int{10}
 }
 func (m *PublishActivityOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -770,7 +770,7 @@ func (m *PublishActivityOp) XXX_Marshal(b []byte, deterministic bool) ([]byte, e
 		return xxx_messageInfo_PublishActivityOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -809,7 +809,7 @@ 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}
+	return fileDescriptor_41f4a519b878ee3b, []int{11}
 }
 func (m *SetStreamReadonlyOp) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -819,7 +819,7 @@ func (m *SetStreamReadonlyOp) XXX_Marshal(b []byte, deterministic bool) ([]byte,
 		return xxx_messageInfo_SetStreamReadonlyOp.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -870,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{12}
+	return fileDescriptor_41f4a519b878ee3b, []int{12}
 }
 func (m *NullableInt64) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -880,7 +880,7 @@ func (m *NullableInt64) XXX_Marshal(b []byte, deterministic bool) ([]byte, error
 		return xxx_messageInfo_NullableInt64.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -917,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{13}
+	return fileDescriptor_41f4a519b878ee3b, []int{13}
 }
 func (m *NullableInt32) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -927,7 +927,7 @@ func (m *NullableInt32) XXX_Marshal(b []byte, deterministic bool) ([]byte, error
 		return xxx_messageInfo_NullableInt32.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -964,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{14}
+	return fileDescriptor_41f4a519b878ee3b, []int{14}
 }
 func (m *NullableBool) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -974,7 +974,7 @@ func (m *NullableBool) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
 		return xxx_messageInfo_NullableBool.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1012,6 +1012,7 @@ type StreamConfig struct {
 	AutoPauseTime                 *NullableInt64 `protobuf:"bytes,9,opt,name=autoPauseTime,proto3" json:"autoPauseTime,omitempty"`
 	AutoPauseDisableIfSubscribers *NullableBool  `protobuf:"bytes,10,opt,name=autoPauseDisableIfSubscribers,proto3" json:"autoPauseDisableIfSubscribers,omitempty"`
 	MinIsr                        *NullableInt32 `protobuf:"bytes,11,opt,name=minIsr,proto3" json:"minIsr,omitempty"`
+	OptimisticConcurrencyControl  *NullableBool  `protobuf:"bytes,12,opt,name=optimisticConcurrencyControl,proto3" json:"optimisticConcurrencyControl,omitempty"`
 	XXX_NoUnkeyedLiteral          struct{}       `json:"-"`
 	XXX_unrecognized              []byte         `json:"-"`
 	XXX_sizecache                 int32          `json:"-"`
@@ -1021,7 +1022,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{15}
+	return fileDescriptor_41f4a519b878ee3b, []int{15}
 }
 func (m *StreamConfig) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1031,7 +1032,7 @@ func (m *StreamConfig) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
 		return xxx_messageInfo_StreamConfig.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1127,6 +1128,13 @@ func (m *StreamConfig) GetMinIsr() *NullableInt32 {
 	return nil
 }
 
+func (m *StreamConfig) GetOptimisticConcurrencyControl() *NullableBool {
+	if m != nil {
+		return m.OptimisticConcurrencyControl
+	}
+	return nil
+}
+
 type Stream struct {
 	Name                 string        `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
 	Subject              string        `protobuf:"bytes,2,opt,name=subject,proto3" json:"subject,omitempty"`
@@ -1142,7 +1150,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{16}
+	return fileDescriptor_41f4a519b878ee3b, []int{16}
 }
 func (m *Stream) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1152,7 +1160,7 @@ func (m *Stream) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 		return xxx_messageInfo_Stream.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1228,7 +1236,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{17}
+	return fileDescriptor_41f4a519b878ee3b, []int{17}
 }
 func (m *Partition) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1238,7 +1246,7 @@ func (m *Partition) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 		return xxx_messageInfo_Partition.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1354,7 +1362,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{18}
+	return fileDescriptor_41f4a519b878ee3b, []int{18}
 }
 func (m *RaftJoinRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1364,7 +1372,7 @@ func (m *RaftJoinRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, err
 		return xxx_messageInfo_RaftJoinRequest.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1409,7 +1417,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{19}
+	return fileDescriptor_41f4a519b878ee3b, []int{19}
 }
 func (m *RaftJoinResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1419,7 +1427,7 @@ func (m *RaftJoinResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, er
 		return xxx_messageInfo_RaftJoinResponse.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1456,7 +1464,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{20}
+	return fileDescriptor_41f4a519b878ee3b, []int{20}
 }
 func (m *MetadataSnapshot) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1466,7 +1474,7 @@ func (m *MetadataSnapshot) XXX_Marshal(b []byte, deterministic bool) ([]byte, er
 		return xxx_messageInfo_MetadataSnapshot.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1505,7 +1513,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{21}
+	return fileDescriptor_41f4a519b878ee3b, []int{21}
 }
 func (m *ReplicationRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1515,7 +1523,7 @@ func (m *ReplicationRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte,
 		return xxx_messageInfo_ReplicationRequest.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1566,7 +1574,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{22}
+	return fileDescriptor_41f4a519b878ee3b, []int{22}
 }
 func (m *LeaderEpochOffsetRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1576,7 +1584,7 @@ func (m *LeaderEpochOffsetRequest) XXX_Marshal(b []byte, deterministic bool) ([]
 		return xxx_messageInfo_LeaderEpochOffsetRequest.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1613,7 +1621,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{23}
+	return fileDescriptor_41f4a519b878ee3b, []int{23}
 }
 func (m *LeaderEpochOffsetResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1623,7 +1631,7 @@ func (m *LeaderEpochOffsetResponse) XXX_Marshal(b []byte, deterministic bool) ([
 		return xxx_messageInfo_LeaderEpochOffsetResponse.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1668,7 +1676,7 @@ 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{24}
+	return fileDescriptor_41f4a519b878ee3b, []int{24}
 }
 func (m *PropagatedRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1678,7 +1686,7 @@ func (m *PropagatedRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, e
 		return xxx_messageInfo_PropagatedRequest.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1772,7 +1780,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{25}
+	return fileDescriptor_41f4a519b878ee3b, []int{25}
 }
 func (m *Error) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1782,7 +1790,7 @@ func (m *Error) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 		return xxx_messageInfo_Error.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1827,7 +1835,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{26}
+	return fileDescriptor_41f4a519b878ee3b, []int{26}
 }
 func (m *PropagatedResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1837,7 +1845,7 @@ func (m *PropagatedResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte,
 		return xxx_messageInfo_PropagatedResponse.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1881,7 +1889,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{27}
+	return fileDescriptor_41f4a519b878ee3b, []int{27}
 }
 func (m *ServerInfoRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1891,7 +1899,7 @@ func (m *ServerInfoRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, e
 		return xxx_messageInfo_ServerInfoRequest.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1930,7 +1938,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{28}
+	return fileDescriptor_41f4a519b878ee3b, []int{28}
 }
 func (m *ServerInfoResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1940,7 +1948,7 @@ func (m *ServerInfoResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte,
 		return xxx_messageInfo_ServerInfoResponse.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -1992,7 +2000,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{29}
+	return fileDescriptor_41f4a519b878ee3b, []int{29}
 }
 func (m *PartitionStatusRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -2002,7 +2010,7 @@ func (m *PartitionStatusRequest) XXX_Marshal(b []byte, deterministic bool) ([]by
 		return xxx_messageInfo_PartitionStatusRequest.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -2047,7 +2055,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{30}
+	return fileDescriptor_41f4a519b878ee3b, []int{30}
 }
 func (m *PartitionStatusResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -2057,7 +2065,7 @@ func (m *PartitionStatusResponse) XXX_Marshal(b []byte, deterministic bool) ([]b
 		return xxx_messageInfo_PartitionStatusResponse.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -2102,7 +2110,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{31}
+	return fileDescriptor_41f4a519b878ee3b, []int{31}
 }
 func (m *PartitionNotification) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -2112,7 +2120,7 @@ func (m *PartitionNotification) XXX_Marshal(b []byte, deterministic bool) ([]byt
 		return xxx_messageInfo_PartitionNotification.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -2159,7 +2167,7 @@ func (m *Cursor) Reset()         { *m = Cursor{} }
 func (m *Cursor) String() string { return proto.CompactTextString(m) }
 func (*Cursor) ProtoMessage()    {}
 func (*Cursor) Descriptor() ([]byte, []int) {
-	return fileDescriptor_7d9410777bf851c3, []int{32}
+	return fileDescriptor_41f4a519b878ee3b, []int{32}
 }
 func (m *Cursor) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -2169,7 +2177,7 @@ func (m *Cursor) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 		return xxx_messageInfo_Cursor.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
+		n, err := m.MarshalTo(b)
 		if err != nil {
 			return nil, err
 		}
@@ -2253,114 +2261,116 @@ func init() {
 	proto.RegisterType((*Cursor)(nil), "protocol.Cursor")
 }
 
-func init() { proto.RegisterFile("server/protocol/internal.proto", fileDescriptor_7d9410777bf851c3) }
-
-var fileDescriptor_7d9410777bf851c3 = []byte{
-	// 1565 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x58, 0xdd, 0x6e, 0x1b, 0xc7,
-	0x15, 0x36, 0x7f, 0x45, 0x1e, 0x4a, 0x34, 0x35, 0xb2, 0x65, 0xd6, 0xb5, 0x05, 0x61, 0x5b, 0x03,
-	0xac, 0xd1, 0xca, 0xa8, 0x5c, 0xb4, 0x68, 0xd1, 0x1a, 0xa5, 0xa4, 0xad, 0xc5, 0x9a, 0x12, 0x85,
-	0x21, 0x5d, 0xd4, 0x40, 0x51, 0x61, 0xc4, 0x1d, 0x52, 0xdb, 0x2e, 0x77, 0x36, 0x33, 0x43, 0x41,
-	0x7e, 0x80, 0xbc, 0x43, 0x90, 0xbb, 0x5c, 0xe5, 0x21, 0x72, 0x99, 0x5c, 0x24, 0x77, 0x79, 0x84,
-	0xc0, 0x79, 0x8d, 0x5c, 0x04, 0x33, 0x3b, 0xfb, 0x4b, 0x51, 0x41, 0xe8, 0x9b, 0x00, 0xb9, 0x9b,
-	0x73, 0xe6, 0x3b, 0x3f, 0x33, 0x7b, 0xe6, 0x3b, 0x33, 0x0b, 0x3b, 0x82, 0xf2, 0x2b, 0xca, 0x9f,
-	0x05, 0x9c, 0x49, 0x36, 0x66, 0xde, 0x33, 0xd7, 0x97, 0x94, 0xfb, 0xc4, 0xdb, 0xd3, 0x1a, 0x54,
-	0x8b, 0x26, 0xac, 0xdf, 0x40, 0x63, 0xa8, 0xb1, 0x43, 0x49, 0x24, 0x45, 0x0f, 0xa1, 0x16, 0x9a,
-	0xf6, 0x8e, 0xda, 0x85, 0xdd, 0x42, 0xa7, 0x8e, 0x63, 0xd9, 0xfa, 0xae, 0x0c, 0x6b, 0x98, 0x4c,
-	0x64, 0x9f, 0x4d, 0xd1, 0x23, 0x28, 0xb2, 0x40, 0x23, 0x9a, 0xfb, 0xeb, 0x7b, 0x91, 0xb7, 0xbd,
-	0x41, 0x80, 0x8b, 0x2c, 0x40, 0x7f, 0x87, 0xe6, 0x98, 0x53, 0x22, 0xe9, 0x50, 0x72, 0x4a, 0x66,
-	0x83, 0xa0, 0x5d, 0xdc, 0x2d, 0x74, 0x1a, 0xfb, 0xed, 0x04, 0x79, 0x98, 0x99, 0xc7, 0x39, 0x3c,
-	0xfa, 0x13, 0x34, 0xc4, 0x25, 0x77, 0xfd, 0xff, 0xf7, 0x86, 0x78, 0x10, 0xb4, 0x4b, 0xda, 0xfc,
-	0x7e, 0x62, 0x3e, 0x4c, 0x26, 0x71, 0x1a, 0xa9, 0x43, 0x5f, 0x12, 0x7f, 0x4a, 0xfb, 0x94, 0x38,
-	0x94, 0x0f, 0x82, 0x76, 0x79, 0x21, 0x74, 0x66, 0x1e, 0xe7, 0xf0, 0x2a, 0x34, 0xbd, 0x0e, 0x88,
-	0xef, 0x84, 0xa1, 0x2b, 0xf9, 0xd0, 0x76, 0x32, 0x89, 0xd3, 0x48, 0x15, 0xda, 0xa1, 0x1e, 0x4d,
-	0xad, 0xba, 0x9a, 0x0f, 0x7d, 0x94, 0x99, 0xc7, 0x39, 0x3c, 0xfa, 0x1b, 0x6c, 0x04, 0x64, 0x2e,
-	0x12, 0x07, 0x6b, 0xda, 0xc1, 0x83, 0xc4, 0xc1, 0x59, 0x7a, 0x1a, 0x67, 0xd1, 0x2a, 0x01, 0x4e,
-	0xc5, 0x7c, 0x96, 0xd8, 0xd7, 0xf2, 0x09, 0xe0, 0xcc, 0x3c, 0xce, 0xe1, 0x51, 0x0f, 0x36, 0x83,
-	0xf9, 0x85, 0xe7, 0x8a, 0xcb, 0xee, 0x58, 0xba, 0x57, 0xae, 0x7c, 0x3b, 0x08, 0xda, 0x75, 0xed,
-	0xe4, 0x97, 0xa9, 0x24, 0xf2, 0x10, 0xbc, 0x68, 0x85, 0x06, 0xb0, 0x25, 0xa8, 0x0c, 0x3d, 0x63,
-	0x4a, 0x1c, 0xe6, 0x7b, 0xca, 0x19, 0x68, 0x67, 0x8f, 0x53, 0x5f, 0x72, 0x11, 0x84, 0x6f, 0xb2,
-	0xb4, 0xfe, 0x02, 0xcd, 0x6c, 0xd1, 0xa0, 0x0e, 0x54, 0x85, 0x1e, 0xeb, 0x42, 0x6c, 0xec, 0xb7,
-	0x52, 0x5e, 0x43, 0x6b, 0x33, 0x6f, 0x7d, 0x5a, 0x80, 0x46, 0xaa, 0x64, 0xd0, 0x76, 0xc6, 0xb2,
-	0x1e, 0xe1, 0xd0, 0x23, 0xa8, 0x07, 0x84, 0x4b, 0x57, 0xba, 0xcc, 0xd7, 0x35, 0x5b, 0xc1, 0x89,
-	0x02, 0x75, 0xe0, 0x2e, 0xa7, 0x81, 0xe7, 0x8e, 0xc9, 0x88, 0x61, 0x3a, 0x63, 0x57, 0x54, 0x17,
-	0x66, 0x1d, 0xe7, 0xd5, 0xca, 0xbf, 0xa7, 0xeb, 0x49, 0x57, 0x5f, 0x1d, 0x1b, 0x09, 0xed, 0x42,
-	0x23, 0x1c, 0xd9, 0x01, 0x1b, 0x5f, 0xea, 0xda, 0x2a, 0xe3, 0xb4, 0xca, 0xfa, 0xa4, 0x00, 0x8d,
-	0x54, 0x85, 0xad, 0x98, 0xa9, 0x05, 0xeb, 0x71, 0x4a, 0x5d, 0xc7, 0x31, 0x69, 0x66, 0x74, 0xef,
-	0x91, 0x63, 0x07, 0x9a, 0xd9, 0x42, 0x5e, 0x96, 0xa5, 0x45, 0x61, 0x23, 0x53, 0xb1, 0x4b, 0x97,
-	0xb3, 0x03, 0x10, 0x67, 0x2f, 0xda, 0xc5, 0xdd, 0x52, 0xa7, 0x82, 0x53, 0x1a, 0xb5, 0xdc, 0xb0,
-	0x54, 0xbb, 0x9e, 0xa7, 0x57, 0x53, 0xc3, 0x89, 0xc2, 0x3a, 0x86, 0x66, 0xb6, 0xb0, 0x57, 0x8d,
-	0x63, 0x7d, 0x5c, 0x50, 0xae, 0x02, 0xc6, 0x65, 0xcc, 0x07, 0xab, 0x7d, 0x81, 0x36, 0xac, 0x99,
-	0xdd, 0x36, 0x9b, 0x1f, 0x89, 0xef, 0xb1, 0xef, 0xff, 0x85, 0x66, 0x96, 0xbb, 0x56, 0xcc, 0x2d,
-	0xc9, 0xa0, 0x94, 0xce, 0xc0, 0xfa, 0x3d, 0x6c, 0x2e, 0x1c, 0x6d, 0xbd, 0xf3, 0x64, 0x22, 0x7b,
-	0xbe, 0x43, 0xaf, 0x75, 0x94, 0x32, 0x4e, 0x14, 0x96, 0x0b, 0x5b, 0x37, 0x1c, 0xe0, 0x95, 0x3f,
-	0xf3, 0x43, 0xa8, 0x71, 0xe3, 0xc5, 0x7c, 0xe5, 0x58, 0xb6, 0x9e, 0xc0, 0xc6, 0xe9, 0xdc, 0xf3,
-	0xc8, 0x85, 0x47, 0x7b, 0xbe, 0xfc, 0xe3, 0x1f, 0xd0, 0x3d, 0xa8, 0x5c, 0x11, 0x6f, 0x4e, 0x75,
-	0x8c, 0x12, 0x0e, 0x85, 0x1c, 0xec, 0xf9, 0x7e, 0x16, 0x56, 0x89, 0x60, 0xbf, 0x86, 0xf5, 0x08,
-	0x76, 0xc0, 0x98, 0x97, 0x45, 0xd5, 0x22, 0xd4, 0x87, 0x55, 0x58, 0x0f, 0x17, 0x77, 0xc8, 0xfc,
-	0x89, 0x3b, 0x45, 0x36, 0x6c, 0x72, 0x2a, 0xa9, 0xaf, 0xd2, 0x3d, 0x21, 0xd7, 0x07, 0x6f, 0x25,
-	0x15, 0x86, 0x7d, 0x52, 0x2c, 0x9d, 0xc9, 0x13, 0x2f, 0x5a, 0xa0, 0x57, 0x70, 0x2f, 0xad, 0x3c,
-	0xa1, 0x42, 0x90, 0x29, 0x15, 0xa6, 0x4d, 0x2e, 0xf5, 0x74, 0xa3, 0x11, 0xea, 0x2a, 0x5a, 0x4a,
-	0xf4, 0xdd, 0x29, 0x35, 0xfd, 0x72, 0xa9, 0x9f, 0x3c, 0x5e, 0xb9, 0x18, 0x7b, 0x94, 0xf8, 0x94,
-	0xf7, 0xd4, 0x45, 0xe1, 0x8a, 0x78, 0xa6, 0x6d, 0x2e, 0x77, 0x91, 0xc3, 0x2b, 0x17, 0x82, 0x4e,
-	0x67, 0xd4, 0x97, 0xf1, 0xbe, 0x54, 0x7e, 0xc0, 0x45, 0x0e, 0xaf, 0xda, 0x5f, 0xa2, 0x52, 0xcb,
-	0xa8, 0xde, 0xee, 0x20, 0x8b, 0x56, 0x9b, 0x3a, 0x66, 0xb3, 0x80, 0x8c, 0x95, 0xe2, 0x25, 0xe3,
-	0x6c, 0x2e, 0x5d, 0x9f, 0x8a, 0xc5, 0x26, 0x9a, 0xa9, 0x0f, 0x7c, 0xa3, 0x11, 0x7a, 0x01, 0x4d,
-	0xa3, 0xb7, 0x7d, 0x85, 0x75, 0x4c, 0x2f, 0xdd, 0x5e, 0x74, 0xa3, 0xea, 0x07, 0xe7, 0xd0, 0x6a,
-	0x2d, 0x64, 0x2e, 0x99, 0x66, 0xbf, 0x91, 0x3b, 0xa3, 0xa6, 0x8b, 0x2e, 0x5f, 0x4b, 0x06, 0x8d,
-	0xfe, 0x03, 0x8f, 0x63, 0xc5, 0x91, 0x2b, 0x34, 0x6e, 0x32, 0x9c, 0x5f, 0x88, 0x31, 0x77, 0x2f,
-	0x28, 0x17, 0xa6, 0x8f, 0x2e, 0xcb, 0xe6, 0x76, 0x63, 0xf4, 0x0c, 0xaa, 0x33, 0xd7, 0xef, 0x09,
-	0xde, 0x6e, 0xdc, 0xbe, 0x37, 0x06, 0x66, 0x7d, 0x55, 0x80, 0x6a, 0x78, 0x0e, 0x10, 0x82, 0xb2,
-	0x4f, 0x66, 0xd4, 0x1c, 0x6c, 0x3d, 0x56, 0x64, 0x27, 0xe6, 0x17, 0xff, 0xa3, 0x63, 0xa9, 0x2b,
-	0xb8, 0x8e, 0x23, 0x11, 0x3d, 0xcf, 0x1c, 0xf8, 0xd2, 0x6e, 0xa9, 0xd3, 0xd8, 0xdf, 0x4a, 0x5f,
-	0x67, 0xcc, 0x5c, 0x86, 0x05, 0xf6, 0xa0, 0x3a, 0xd6, 0xc7, 0xcd, 0x14, 0xe1, 0x76, 0xbe, 0xaf,
-	0x87, 0x87, 0x11, 0x1b, 0x14, 0xfa, 0x2d, 0x6c, 0xea, 0xeb, 0xa3, 0xcb, 0x7c, 0xb5, 0x79, 0x42,
-	0x92, 0x59, 0x78, 0x6f, 0x2b, 0xe1, 0xc5, 0x09, 0xeb, 0xf3, 0x22, 0xd4, 0xcf, 0xd2, 0x3c, 0x1d,
-	0xa5, 0x5e, 0xc8, 0xa6, 0x9e, 0x70, 0x58, 0x31, 0xc3, 0x61, 0x4d, 0x28, 0xba, 0x61, 0x47, 0xad,
-	0xe0, 0xa2, 0xeb, 0x28, 0xe6, 0x98, 0x72, 0x36, 0x0f, 0x0c, 0x9d, 0x87, 0x82, 0xca, 0xc9, 0x10,
-	0xbe, 0x0a, 0xf3, 0x0f, 0x32, 0x96, 0x8c, 0xeb, 0x9c, 0x2a, 0x78, 0x71, 0x22, 0xe4, 0x3d, 0xad,
-	0x14, 0xed, 0xea, 0x6e, 0x49, 0x5d, 0xbb, 0x23, 0x39, 0xc5, 0xd6, 0x6b, 0x99, 0x7e, 0xd1, 0x82,
-	0x92, 0x2b, 0x78, 0xbb, 0xa6, 0xe1, 0x6a, 0x98, 0xef, 0x20, 0xf5, 0x85, 0x0e, 0xa2, 0x72, 0xa5,
-	0x7a, 0x0e, 0xf4, 0x5c, 0x28, 0xa8, 0x08, 0xfa, 0x22, 0xe9, 0xe8, 0x72, 0xa8, 0x61, 0x23, 0x65,
-	0xd8, 0x78, 0x3d, 0xc7, 0xc6, 0x36, 0xdc, 0x55, 0x6f, 0x81, 0x7f, 0x32, 0xd7, 0xc7, 0xf4, 0x83,
-	0x39, 0x15, 0x7a, 0xc3, 0x7c, 0xe6, 0xd0, 0xf8, 0xe5, 0x60, 0x24, 0xe5, 0x46, 0x8d, 0xba, 0x8e,
-	0xc3, 0xcd, 0x56, 0xc6, 0xb2, 0xd5, 0x81, 0x56, 0xe2, 0x46, 0x04, 0xcc, 0x17, 0x54, 0x27, 0xc9,
-	0x39, 0xe3, 0xc6, 0x4d, 0x28, 0x58, 0x2f, 0xa0, 0x75, 0x42, 0x25, 0x71, 0x88, 0x24, 0x43, 0x9f,
-	0x04, 0xe2, 0x92, 0x49, 0xf4, 0x14, 0xd6, 0xc2, 0x8f, 0xa2, 0x38, 0xb8, 0x74, 0xe3, 0x0d, 0x30,
-	0x02, 0x58, 0x1e, 0x20, 0x9c, 0xec, 0x7b, 0x94, 0xb3, 0xbe, 0x57, 0x68, 0x6d, 0x9c, 0x76, 0xa2,
-	0x50, 0x2b, 0x62, 0x93, 0x89, 0xa0, 0x61, 0x59, 0x97, 0xb0, 0x91, 0xf2, 0x1b, 0x5d, 0x5a, 0x6c,
-	0xd5, 0x7f, 0x85, 0x76, 0x3f, 0x11, 0x07, 0xda, 0x2c, 0x8a, 0x99, 0xb3, 0x2e, 0x2c, 0x5a, 0xff,
-	0x19, 0x7e, 0x71, 0x83, 0xb5, 0xd9, 0x9e, 0x47, 0x50, 0xa7, 0xbe, 0x13, 0x2a, 0x4d, 0xeb, 0x4b,
-	0x14, 0xd6, 0x67, 0x65, 0xd8, 0x3c, 0xe3, 0x2c, 0x20, 0x53, 0x22, 0xa9, 0x93, 0x2c, 0xf3, 0xa7,
-	0xfb, 0x5c, 0xe3, 0x99, 0xeb, 0xd6, 0xe2, 0x73, 0x2d, 0x7b, 0x1d, 0xc3, 0x39, 0xfc, 0xcf, 0xfa,
-	0xb9, 0xb6, 0xe4, 0x8d, 0x55, 0x5f, 0xf9, 0x8d, 0xf5, 0x3b, 0xa8, 0xd8, 0xea, 0xb4, 0x29, 0x96,
-	0x1f, 0x33, 0x27, 0x64, 0xf9, 0x0d, 0xac, 0xc7, 0x8a, 0x70, 0x66, 0x62, 0x6a, 0x8e, 0xb0, 0x1a,
-	0x5a, 0x6f, 0x00, 0xa5, 0x6b, 0x2d, 0x2e, 0xd0, 0xdb, 0x8a, 0xed, 0x49, 0x74, 0xba, 0xc3, 0x1a,
-	0xbb, 0x9b, 0xfa, 0x52, 0x4a, 0x1d, 0x1d, 0xf7, 0x5f, 0xc1, 0x66, 0xf8, 0x5f, 0xa2, 0xe7, 0x4f,
-	0x58, 0x54, 0xc6, 0x21, 0xf5, 0x86, 0xc7, 0xb4, 0xe8, 0x3a, 0x56, 0x1f, 0x50, 0x1a, 0x64, 0xe2,
-	0xe7, 0x50, 0x6a, 0x2d, 0x97, 0x4c, 0x44, 0xad, 0x49, 0x8f, 0x95, 0x4e, 0x55, 0x91, 0xa1, 0x71,
-	0x3d, 0xb6, 0x4e, 0x61, 0x3b, 0xee, 0x0b, 0x43, 0x49, 0xe4, 0x5c, 0xa4, 0x98, 0xed, 0xc7, 0x5f,
-	0xb3, 0xad, 0x13, 0x78, 0xb0, 0xe0, 0xcf, 0xa4, 0xb8, 0x0d, 0x55, 0x7a, 0xed, 0x0a, 0x29, 0xcc,
-	0x75, 0xd3, 0x48, 0x8a, 0x2a, 0x5d, 0x11, 0x96, 0xb6, 0xf6, 0x57, 0xc3, 0xb1, 0x6c, 0x9d, 0xc0,
-	0xfd, 0xd8, 0xdd, 0x29, 0x93, 0xee, 0xc4, 0x50, 0xd9, 0x8a, 0xd9, 0x71, 0xa8, 0x1e, 0xce, 0xb9,
-	0x60, 0x7c, 0xc5, 0x47, 0xc4, 0x43, 0xa8, 0x8d, 0xb5, 0x7d, 0x2f, 0x7a, 0x5e, 0xc6, 0x72, 0x8a,
-	0x37, 0xcb, 0x69, 0xde, 0x7c, 0xfa, 0x45, 0x01, 0x8a, 0x83, 0x00, 0x6d, 0xc2, 0xc6, 0x21, 0xb6,
-	0xbb, 0x23, 0xfb, 0x7c, 0x38, 0xc2, 0x76, 0xf7, 0xa4, 0x75, 0x07, 0x35, 0x01, 0x86, 0xc7, 0xb8,
-	0x77, 0xfa, 0xea, 0xbc, 0x37, 0xc4, 0xad, 0x82, 0x82, 0x60, 0xfb, 0x6c, 0x80, 0x47, 0xe7, 0x7d,
-	0xbb, 0x7b, 0x64, 0xe3, 0x56, 0x51, 0x5b, 0x1d, 0x77, 0x4f, 0x5f, 0xda, 0x91, 0xaa, 0xa4, 0xac,
-	0xec, 0x7f, 0x9f, 0x75, 0x4f, 0x8f, 0xb4, 0x55, 0x59, 0x41, 0x8e, 0xec, 0xbe, 0x9d, 0x38, 0xae,
-	0xa0, 0x16, 0xac, 0x9f, 0x75, 0x5f, 0x0f, 0x63, 0x4d, 0x35, 0x74, 0x3d, 0x7c, 0x7d, 0x12, 0xab,
-	0xd6, 0xd0, 0x3d, 0x68, 0x9d, 0xbd, 0x3e, 0xe8, 0xf7, 0x86, 0xc7, 0xe7, 0xdd, 0xc3, 0x51, 0xef,
-	0x5f, 0xbd, 0xd1, 0x9b, 0x56, 0x0d, 0x3d, 0x80, 0xad, 0xa1, 0x3d, 0x32, 0xa8, 0x73, 0x6c, 0x77,
-	0x8f, 0x06, 0xa7, 0xfd, 0x37, 0xad, 0xfa, 0x41, 0xeb, 0xcb, 0x77, 0x3b, 0x85, 0xaf, 0xdf, 0xed,
-	0x14, 0xbe, 0x79, 0xb7, 0x53, 0xf8, 0xe8, 0xdb, 0x9d, 0x3b, 0x17, 0x55, 0x5d, 0xc4, 0xcf, 0xbf,
-	0x0f, 0x00, 0x00, 0xff, 0xff, 0x3d, 0xa9, 0x32, 0xf7, 0x78, 0x13, 0x00, 0x00,
+func init() { proto.RegisterFile("internal.proto", fileDescriptor_41f4a519b878ee3b) }
+
+var fileDescriptor_41f4a519b878ee3b = []byte{
+	// 1590 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe4, 0x58, 0x5f, 0x6f, 0x1b, 0x37,
+	0x12, 0x8f, 0xfe, 0x5a, 0x1a, 0xd9, 0x8a, 0x4c, 0x27, 0x8e, 0x2e, 0xe7, 0x18, 0xc6, 0xde, 0x05,
+	0xd0, 0x05, 0x77, 0x3e, 0x9c, 0x73, 0xb8, 0x43, 0x8b, 0x36, 0xa8, 0x6c, 0x6f, 0x63, 0x35, 0xb2,
+	0x65, 0x50, 0x4a, 0xd1, 0x14, 0x45, 0x8d, 0xf5, 0x2e, 0x25, 0x6f, 0xbb, 0x5a, 0x6e, 0x49, 0xca,
+	0x70, 0xbe, 0x49, 0xd1, 0xb7, 0x3e, 0xf5, 0x43, 0xf4, 0xb1, 0x7d, 0x68, 0xdf, 0xfa, 0x11, 0xda,
+	0xf4, 0x6b, 0xf4, 0xa1, 0x20, 0x97, 0xfb, 0x57, 0x96, 0x8b, 0x2a, 0x2f, 0x05, 0xfa, 0x24, 0xce,
+	0xf0, 0x37, 0x3f, 0x0e, 0xb9, 0xc3, 0x99, 0xa1, 0xa0, 0xe9, 0xfa, 0x82, 0x30, 0xdf, 0xf2, 0x76,
+	0x03, 0x46, 0x05, 0x45, 0x35, 0xf5, 0x63, 0x53, 0xcf, 0xf8, 0x07, 0x34, 0x86, 0x84, 0x5d, 0x12,
+	0x36, 0x14, 0x96, 0x20, 0xe8, 0x3e, 0xd4, 0xb8, 0x12, 0x7b, 0x87, 0xed, 0xc2, 0x4e, 0xa1, 0x53,
+	0xc7, 0xb1, 0x6c, 0xfc, 0x52, 0x86, 0x15, 0x6c, 0x8d, 0x45, 0x9f, 0x4e, 0xd0, 0x16, 0x14, 0x69,
+	0xa0, 0x10, 0xcd, 0xbd, 0xd5, 0xdd, 0x88, 0x6d, 0x77, 0x10, 0xe0, 0x22, 0x0d, 0xd0, 0x3b, 0xd0,
+	0xb4, 0x19, 0xb1, 0x04, 0x19, 0x0a, 0x46, 0xac, 0xe9, 0x20, 0x68, 0x17, 0x77, 0x0a, 0x9d, 0xc6,
+	0x5e, 0x3b, 0x41, 0x1e, 0x64, 0xe6, 0x71, 0x0e, 0x8f, 0xfe, 0x0f, 0x0d, 0x7e, 0xc1, 0x5c, 0xff,
+	0xd3, 0xde, 0x10, 0x0f, 0x82, 0x76, 0x49, 0x99, 0xdf, 0x4d, 0xcc, 0x87, 0xc9, 0x24, 0x4e, 0x23,
+	0xd5, 0xd2, 0x17, 0x96, 0x3f, 0x21, 0x7d, 0x62, 0x39, 0x84, 0x0d, 0x82, 0x76, 0x79, 0x6e, 0xe9,
+	0xcc, 0x3c, 0xce, 0xe1, 0xe5, 0xd2, 0xe4, 0x2a, 0xb0, 0x7c, 0x27, 0x5c, 0xba, 0x92, 0x5f, 0xda,
+	0x4c, 0x26, 0x71, 0x1a, 0x29, 0x97, 0x76, 0x88, 0x47, 0x52, 0xbb, 0xae, 0xe6, 0x97, 0x3e, 0xcc,
+	0xcc, 0xe3, 0x1c, 0x1e, 0xbd, 0x0d, 0x6b, 0x81, 0x35, 0xe3, 0x09, 0xc1, 0x8a, 0x22, 0xb8, 0x97,
+	0x10, 0x9c, 0xa6, 0xa7, 0x71, 0x16, 0x2d, 0x1d, 0x60, 0x84, 0xcf, 0xa6, 0x89, 0x7d, 0x2d, 0xef,
+	0x00, 0xce, 0xcc, 0xe3, 0x1c, 0x1e, 0xf5, 0x60, 0x3d, 0x98, 0x9d, 0x7b, 0x2e, 0xbf, 0xe8, 0xda,
+	0xc2, 0xbd, 0x74, 0xc5, 0xcb, 0x41, 0xd0, 0xae, 0x2b, 0x92, 0xbf, 0xa6, 0x9c, 0xc8, 0x43, 0xf0,
+	0xbc, 0x15, 0x1a, 0xc0, 0x06, 0x27, 0x22, 0x64, 0xc6, 0xc4, 0x72, 0xa8, 0xef, 0x49, 0x32, 0x50,
+	0x64, 0x0f, 0x52, 0x5f, 0x72, 0x1e, 0x84, 0xaf, 0xb3, 0x34, 0xde, 0x84, 0x66, 0x36, 0x68, 0x50,
+	0x07, 0xaa, 0x5c, 0x8d, 0x55, 0x20, 0x36, 0xf6, 0x5a, 0x29, 0xd6, 0xd0, 0x5a, 0xcf, 0x1b, 0x5f,
+	0x15, 0xa0, 0x91, 0x0a, 0x19, 0xb4, 0x99, 0xb1, 0xac, 0x47, 0x38, 0xb4, 0x05, 0xf5, 0xc0, 0x62,
+	0xc2, 0x15, 0x2e, 0xf5, 0x55, 0xcc, 0x56, 0x70, 0xa2, 0x40, 0x1d, 0xb8, 0xcd, 0x48, 0xe0, 0xb9,
+	0xb6, 0x35, 0xa2, 0x98, 0x4c, 0xe9, 0x25, 0x51, 0x81, 0x59, 0xc7, 0x79, 0xb5, 0xe4, 0xf7, 0x54,
+	0x3c, 0xa9, 0xe8, 0xab, 0x63, 0x2d, 0xa1, 0x1d, 0x68, 0x84, 0x23, 0x33, 0xa0, 0xf6, 0x85, 0x8a,
+	0xad, 0x32, 0x4e, 0xab, 0x8c, 0x2f, 0x0b, 0xd0, 0x48, 0x45, 0xd8, 0x92, 0x9e, 0x1a, 0xb0, 0x1a,
+	0xbb, 0xd4, 0x75, 0x1c, 0xed, 0x66, 0x46, 0xf7, 0x1a, 0x3e, 0x76, 0xa0, 0x99, 0x0d, 0xe4, 0x45,
+	0x5e, 0x1a, 0x04, 0xd6, 0x32, 0x11, 0xbb, 0x70, 0x3b, 0xdb, 0x00, 0xb1, 0xf7, 0xbc, 0x5d, 0xdc,
+	0x29, 0x75, 0x2a, 0x38, 0xa5, 0x91, 0xdb, 0x0d, 0x43, 0xb5, 0xeb, 0x79, 0x6a, 0x37, 0x35, 0x9c,
+	0x28, 0x8c, 0x23, 0x68, 0x66, 0x03, 0x7b, 0xd9, 0x75, 0x8c, 0x2f, 0x0a, 0x92, 0x2a, 0xa0, 0x4c,
+	0xc4, 0xf9, 0x60, 0xb9, 0x2f, 0xd0, 0x86, 0x15, 0x7d, 0xda, 0xfa, 0xf0, 0x23, 0xf1, 0x35, 0xce,
+	0xfd, 0x63, 0x68, 0x66, 0x73, 0xd7, 0x92, 0xbe, 0x25, 0x1e, 0x94, 0xd2, 0x1e, 0x18, 0xff, 0x81,
+	0xf5, 0xb9, 0xab, 0xad, 0x4e, 0xde, 0x1a, 0x8b, 0x9e, 0xef, 0x90, 0x2b, 0xb5, 0x4a, 0x19, 0x27,
+	0x0a, 0xc3, 0x85, 0x8d, 0x6b, 0x2e, 0xf0, 0xd2, 0x9f, 0xf9, 0x3e, 0xd4, 0x98, 0x66, 0xd1, 0x5f,
+	0x39, 0x96, 0x8d, 0x87, 0xb0, 0x76, 0x32, 0xf3, 0x3c, 0xeb, 0xdc, 0x23, 0x3d, 0x5f, 0xfc, 0xef,
+	0xbf, 0xe8, 0x0e, 0x54, 0x2e, 0x2d, 0x6f, 0x46, 0xd4, 0x1a, 0x25, 0x1c, 0x0a, 0x39, 0xd8, 0xe3,
+	0xbd, 0x2c, 0xac, 0x12, 0xc1, 0xfe, 0x0e, 0xab, 0x11, 0x6c, 0x9f, 0x52, 0x2f, 0x8b, 0xaa, 0x45,
+	0xa8, 0x9f, 0xaa, 0xb0, 0x1a, 0x6e, 0xee, 0x80, 0xfa, 0x63, 0x77, 0x82, 0x4c, 0x58, 0x67, 0x44,
+	0x10, 0x5f, 0xba, 0x7b, 0x6c, 0x5d, 0xed, 0xbf, 0x14, 0x84, 0xeb, 0xec, 0x93, 0xca, 0xd2, 0x19,
+	0x3f, 0xf1, 0xbc, 0x05, 0x7a, 0x06, 0x77, 0xd2, 0xca, 0x63, 0xc2, 0xb9, 0x35, 0x21, 0x5c, 0x97,
+	0xc9, 0x85, 0x4c, 0xd7, 0x1a, 0xa1, 0xae, 0x4c, 0x4b, 0x89, 0xbe, 0x3b, 0x21, 0xba, 0x5e, 0x2e,
+	0xe4, 0xc9, 0xe3, 0x25, 0x85, 0xed, 0x11, 0xcb, 0x27, 0xac, 0x27, 0x1b, 0x85, 0x4b, 0xcb, 0xd3,
+	0x65, 0x73, 0x31, 0x45, 0x0e, 0x2f, 0x29, 0x38, 0x99, 0x4c, 0x89, 0x2f, 0xe2, 0x73, 0xa9, 0xfc,
+	0x06, 0x45, 0x0e, 0x2f, 0xcb, 0x5f, 0xa2, 0x92, 0xdb, 0xa8, 0xde, 0x4c, 0x90, 0x45, 0xcb, 0x43,
+	0xb5, 0xe9, 0x34, 0xb0, 0x6c, 0xa9, 0x78, 0x4a, 0x19, 0x9d, 0x09, 0xd7, 0x27, 0x7c, 0xbe, 0x88,
+	0x66, 0xe2, 0x03, 0x5f, 0x6b, 0x84, 0x9e, 0x40, 0x53, 0xeb, 0x4d, 0x5f, 0x62, 0x1d, 0x5d, 0x4b,
+	0x37, 0xe7, 0x69, 0x64, 0xfc, 0xe0, 0x1c, 0x5a, 0xee, 0xc5, 0x9a, 0x09, 0xaa, 0xb2, 0xdf, 0xc8,
+	0x9d, 0x12, 0x5d, 0x45, 0x17, 0xef, 0x25, 0x83, 0x46, 0x1f, 0xc1, 0x83, 0x58, 0x71, 0xe8, 0x72,
+	0x85, 0x1b, 0x0f, 0x67, 0xe7, 0xdc, 0x66, 0xee, 0x39, 0x61, 0x5c, 0xd7, 0xd1, 0x45, 0xde, 0xdc,
+	0x6c, 0x8c, 0xfe, 0x0d, 0xd5, 0xa9, 0xeb, 0xf7, 0x38, 0x6b, 0x37, 0x6e, 0x3e, 0x1b, 0x0d, 0x43,
+	0x1f, 0xc2, 0x16, 0x0d, 0x84, 0x3b, 0x75, 0xb9, 0x70, 0xed, 0x03, 0xea, 0xdb, 0x33, 0xc6, 0x88,
+	0x6f, 0xbf, 0x3c, 0xa0, 0xbe, 0x60, 0xd4, 0x6b, 0xaf, 0xde, 0xe8, 0xcd, 0x8d, 0xb6, 0xc6, 0xf7,
+	0x05, 0xa8, 0x86, 0x77, 0x0c, 0x21, 0x28, 0xfb, 0xd6, 0x94, 0xe8, 0xa4, 0xa1, 0xc6, 0x32, 0x91,
+	0xf2, 0xd9, 0xf9, 0x27, 0xc4, 0x16, 0xea, 0x76, 0xd4, 0x71, 0x24, 0xa2, 0xc7, 0x99, 0x64, 0x52,
+	0xda, 0x29, 0x75, 0x1a, 0x7b, 0x1b, 0xe9, 0x56, 0x49, 0xcf, 0x65, 0x32, 0xcc, 0x2e, 0x54, 0x6d,
+	0x75, 0x95, 0x75, 0x80, 0x6f, 0xe6, 0x7b, 0x86, 0xf0, 0xa2, 0x63, 0x8d, 0x42, 0xff, 0x84, 0x75,
+	0xd5, 0x9a, 0xba, 0xd4, 0x97, 0x1f, 0x86, 0x0b, 0x6b, 0x1a, 0xf6, 0x84, 0x25, 0x3c, 0x3f, 0x61,
+	0x7c, 0x53, 0x84, 0xfa, 0x69, 0xba, 0x06, 0x44, 0xae, 0x17, 0xb2, 0xae, 0x27, 0xf9, 0xb1, 0x98,
+	0xc9, 0x8f, 0x4d, 0x28, 0xba, 0x61, 0xb5, 0xae, 0xe0, 0xa2, 0xeb, 0xc8, 0xac, 0x34, 0x61, 0x74,
+	0x16, 0xe8, 0x52, 0x11, 0x0a, 0xd2, 0x27, 0x5d, 0x4c, 0xe4, 0x32, 0xef, 0x5a, 0xb6, 0xa0, 0x4c,
+	0xf9, 0x54, 0xc1, 0xf3, 0x13, 0x61, 0x4e, 0x55, 0x4a, 0xde, 0xae, 0xee, 0x94, 0x64, 0x4b, 0x1f,
+	0xc9, 0xa9, 0x4a, 0xb0, 0x92, 0xa9, 0x45, 0x2d, 0x28, 0xb9, 0x9c, 0xb5, 0x6b, 0x0a, 0x2e, 0x87,
+	0xf9, 0xea, 0x54, 0x9f, 0xab, 0x4e, 0xd2, 0x57, 0xa2, 0xe6, 0x40, 0xcd, 0x85, 0x82, 0x5c, 0x41,
+	0x35, 0xa9, 0x8e, 0x0a, 0xb5, 0x1a, 0xd6, 0x52, 0x26, 0xd3, 0xaf, 0xe6, 0x32, 0xbd, 0x09, 0xb7,
+	0xe5, 0x3b, 0xe3, 0x3d, 0xea, 0xfa, 0x98, 0x7c, 0x36, 0x23, 0x5c, 0x1d, 0x98, 0x4f, 0x1d, 0x12,
+	0xbf, 0x4a, 0xb4, 0x24, 0x69, 0xe4, 0xa8, 0xeb, 0x38, 0x4c, 0x1f, 0x65, 0x2c, 0x1b, 0x1d, 0x68,
+	0x25, 0x34, 0x3c, 0xa0, 0x3e, 0x27, 0xca, 0x49, 0xc6, 0x28, 0xd3, 0x34, 0xa1, 0x60, 0x3c, 0x81,
+	0xd6, 0x31, 0x11, 0x96, 0x63, 0x09, 0x6b, 0xe8, 0x5b, 0x01, 0xbf, 0xa0, 0x02, 0x3d, 0x82, 0x95,
+	0xf0, 0xa3, 0xc8, 0xfc, 0x5e, 0xba, 0xb6, 0xbb, 0x8c, 0x00, 0x86, 0x07, 0x08, 0x27, 0xe7, 0x1e,
+	0xf9, 0xac, 0x7a, 0x16, 0xa5, 0x8d, 0xdd, 0x4e, 0x14, 0x72, 0x47, 0x74, 0x3c, 0xe6, 0x24, 0x0c,
+	0xeb, 0x12, 0xd6, 0x52, 0xfe, 0xa0, 0x4b, 0xf3, 0x6d, 0xc0, 0x5b, 0xd0, 0xee, 0x27, 0xe2, 0x40,
+	0x99, 0x45, 0x6b, 0xe6, 0xac, 0x0b, 0xf3, 0xd6, 0x6f, 0xc0, 0x5f, 0xae, 0xb1, 0xd6, 0xc7, 0xb3,
+	0x05, 0x75, 0xe2, 0x3b, 0xa1, 0x52, 0x97, 0xd5, 0x44, 0x61, 0x7c, 0x5d, 0x86, 0xf5, 0x53, 0x46,
+	0x03, 0x6b, 0x62, 0x09, 0xe2, 0x24, 0xdb, 0xfc, 0xe3, 0x3e, 0x05, 0x59, 0xa6, 0x95, 0x9b, 0x7f,
+	0x0a, 0x66, 0x5b, 0x3d, 0x9c, 0xc3, 0xff, 0xa9, 0x9f, 0x82, 0x0b, 0xde, 0x6f, 0xf5, 0xa5, 0xdf,
+	0x6f, 0xff, 0x82, 0x8a, 0x29, 0x6f, 0x9b, 0xcc, 0xf2, 0x36, 0x75, 0xc2, 0x2c, 0xbf, 0x86, 0xd5,
+	0x58, 0x26, 0x9c, 0x29, 0x9f, 0xe8, 0x2b, 0x2c, 0x87, 0xc6, 0x0b, 0x40, 0xe9, 0x58, 0x8b, 0x03,
+	0xf4, 0xa6, 0x60, 0x7b, 0x18, 0xdd, 0xee, 0x30, 0xc6, 0x6e, 0xa7, 0xbe, 0x94, 0x54, 0x47, 0xd7,
+	0xfd, 0x6f, 0xb0, 0x1e, 0xfe, 0xe7, 0xd1, 0xf3, 0xc7, 0x34, 0x0a, 0xe3, 0x30, 0xf5, 0x86, 0xd7,
+	0xb4, 0xe8, 0x3a, 0x46, 0x1f, 0x50, 0x1a, 0xa4, 0xd7, 0xcf, 0xa1, 0xe4, 0x5e, 0x2e, 0x28, 0x8f,
+	0x4a, 0x93, 0x1a, 0x4b, 0x9d, 0x8c, 0x22, 0x9d, 0xc6, 0xd5, 0xd8, 0x38, 0x81, 0xcd, 0xb8, 0x2e,
+	0x0c, 0x85, 0x25, 0x66, 0x3c, 0x95, 0xd9, 0x7e, 0x7f, 0x0b, 0x6f, 0x1c, 0xc3, 0xbd, 0x39, 0x3e,
+	0xed, 0xe2, 0x26, 0x54, 0xc9, 0x95, 0xcb, 0x05, 0xd7, 0xad, 0xac, 0x96, 0x64, 0xaa, 0x74, 0x79,
+	0x18, 0xda, 0x8a, 0xaf, 0x86, 0x63, 0xd9, 0x38, 0x86, 0xbb, 0x31, 0xdd, 0x09, 0x15, 0xee, 0x58,
+	0xa7, 0xb2, 0x25, 0xbd, 0x63, 0x50, 0x3d, 0x98, 0x31, 0x4e, 0xd9, 0x92, 0x0f, 0x94, 0xfb, 0x50,
+	0xb3, 0x95, 0x7d, 0x2f, 0x7a, 0xba, 0xc6, 0x72, 0x2a, 0x6f, 0x96, 0xd3, 0x79, 0xf3, 0xd1, 0xb7,
+	0x05, 0x28, 0x0e, 0x02, 0xb4, 0x0e, 0x6b, 0x07, 0xd8, 0xec, 0x8e, 0xcc, 0xb3, 0xe1, 0x08, 0x9b,
+	0xdd, 0xe3, 0xd6, 0x2d, 0xd4, 0x04, 0x18, 0x1e, 0xe1, 0xde, 0xc9, 0xb3, 0xb3, 0xde, 0x10, 0xb7,
+	0x0a, 0x12, 0x82, 0xcd, 0xd3, 0x01, 0x1e, 0x9d, 0xf5, 0xcd, 0xee, 0xa1, 0x89, 0x5b, 0x45, 0x65,
+	0x75, 0xd4, 0x3d, 0x79, 0x6a, 0x46, 0xaa, 0x92, 0xb4, 0x32, 0x3f, 0x38, 0xed, 0x9e, 0x1c, 0x2a,
+	0xab, 0xb2, 0x84, 0x1c, 0x9a, 0x7d, 0x33, 0x21, 0xae, 0xa0, 0x16, 0xac, 0x9e, 0x76, 0x9f, 0x0f,
+	0x63, 0x4d, 0x35, 0xa4, 0x1e, 0x3e, 0x3f, 0x8e, 0x55, 0x2b, 0xe8, 0x0e, 0xb4, 0x4e, 0x9f, 0xef,
+	0xf7, 0x7b, 0xc3, 0xa3, 0xb3, 0xee, 0xc1, 0xa8, 0xf7, 0x7e, 0x6f, 0xf4, 0xa2, 0x55, 0x43, 0xf7,
+	0x60, 0x63, 0x68, 0x8e, 0x34, 0xea, 0x0c, 0x9b, 0xdd, 0xc3, 0xc1, 0x49, 0xff, 0x45, 0xab, 0xbe,
+	0xdf, 0xfa, 0xee, 0xd5, 0x76, 0xe1, 0x87, 0x57, 0xdb, 0x85, 0x1f, 0x5f, 0x6d, 0x17, 0x3e, 0xff,
+	0x79, 0xfb, 0xd6, 0x79, 0x55, 0x05, 0xf1, 0xe3, 0x5f, 0x03, 0x00, 0x00, 0xff, 0xff, 0x3b, 0x21,
+	0x94, 0x0a, 0xc4, 0x13, 0x00, 0x00,
 }
 
 func (m *ServerState) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2368,33 +2378,26 @@ func (m *ServerState) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ServerState) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ServerState) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if len(m.ServerID) > 0 {
-		i -= len(m.ServerID)
-		copy(dAtA[i:], m.ServerID)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.ServerID)))
-		i--
 		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.ServerID)))
+		i += copy(dAtA[i:], m.ServerID)
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *RaftLog) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2402,139 +2405,115 @@ func (m *RaftLog) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *RaftLog) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *RaftLog) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if m.Op != 0 {
+		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Op))
 	}
-	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))
+	if m.CreateStreamOp != nil {
+		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.CreateStreamOp.Size()))
+		n1, err1 := m.CreateStreamOp.MarshalTo(dAtA[i:])
+		if err1 != nil {
+			return 0, err1
 		}
-		i--
-		dAtA[i] = 0x52
+		i += n1
 	}
-	if m.PublishActivityOp != nil {
-		{
-			size, err := m.PublishActivityOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ShrinkISROp != nil {
+		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ShrinkISROp.Size()))
+		n2, err2 := m.ShrinkISROp.MarshalTo(dAtA[i:])
+		if err2 != nil {
+			return 0, err2
 		}
-		i--
-		dAtA[i] = 0x4a
+		i += n2
 	}
-	if m.ResumeStreamOp != nil {
-		{
-			size, err := m.ResumeStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ChangeLeaderOp != nil {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ChangeLeaderOp.Size()))
+		n3, err3 := m.ChangeLeaderOp.MarshalTo(dAtA[i:])
+		if err3 != nil {
+			return 0, err3
 		}
-		i--
-		dAtA[i] = 0x42
+		i += n3
 	}
-	if m.PauseStreamOp != nil {
-		{
-			size, err := m.PauseStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ExpandISROp != nil {
+		dAtA[i] = 0x2a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ExpandISROp.Size()))
+		n4, err4 := m.ExpandISROp.MarshalTo(dAtA[i:])
+		if err4 != nil {
+			return 0, err4
 		}
-		i--
-		dAtA[i] = 0x3a
+		i += n4
 	}
 	if m.DeleteStreamOp != nil {
-		{
-			size, err := m.DeleteStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
-		}
-		i--
 		dAtA[i] = 0x32
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.DeleteStreamOp.Size()))
+		n5, err5 := m.DeleteStreamOp.MarshalTo(dAtA[i:])
+		if err5 != nil {
+			return 0, err5
+		}
+		i += n5
 	}
-	if m.ExpandISROp != nil {
-		{
-			size, err := m.ExpandISROp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.PauseStreamOp != nil {
+		dAtA[i] = 0x3a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.PauseStreamOp.Size()))
+		n6, err6 := m.PauseStreamOp.MarshalTo(dAtA[i:])
+		if err6 != nil {
+			return 0, err6
 		}
-		i--
-		dAtA[i] = 0x2a
+		i += n6
 	}
-	if m.ChangeLeaderOp != nil {
-		{
-			size, err := m.ChangeLeaderOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ResumeStreamOp != nil {
+		dAtA[i] = 0x42
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ResumeStreamOp.Size()))
+		n7, err7 := m.ResumeStreamOp.MarshalTo(dAtA[i:])
+		if err7 != nil {
+			return 0, err7
 		}
-		i--
-		dAtA[i] = 0x22
+		i += n7
 	}
-	if m.ShrinkISROp != nil {
-		{
-			size, err := m.ShrinkISROp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.PublishActivityOp != nil {
+		dAtA[i] = 0x4a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.PublishActivityOp.Size()))
+		n8, err8 := m.PublishActivityOp.MarshalTo(dAtA[i:])
+		if err8 != nil {
+			return 0, err8
 		}
-		i--
-		dAtA[i] = 0x1a
+		i += n8
 	}
-	if m.CreateStreamOp != nil {
-		{
-			size, err := m.CreateStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.SetStreamReadonlyOp != nil {
+		dAtA[i] = 0x52
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.SetStreamReadonlyOp.Size()))
+		n9, err9 := m.SetStreamReadonlyOp.MarshalTo(dAtA[i:])
+		if err9 != nil {
+			return 0, err9
 		}
-		i--
-		dAtA[i] = 0x12
+		i += n9
 	}
-	if m.Op != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Op))
-		i--
-		dAtA[i] = 0x8
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *CreateStreamOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2542,38 +2521,30 @@ func (m *CreateStreamOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *CreateStreamOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *CreateStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.Stream != nil {
-		{
-			size, err := m.Stream.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
-		}
-		i--
 		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Stream.Size()))
+		n10, err10 := m.Stream.MarshalTo(dAtA[i:])
+		if err10 != nil {
+			return 0, err10
+		}
+		i += n10
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *ShrinkISROp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2581,57 +2552,48 @@ func (m *ShrinkISROp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ShrinkISROp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ShrinkISROp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.LeaderEpoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
-		i--
-		dAtA[i] = 0x28
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
-	if len(m.Leader) > 0 {
-		i -= len(m.Leader)
-		copy(dAtA[i:], m.Leader)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
-		i--
-		dAtA[i] = 0x22
+	if m.Partition != 0 {
+		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
 	if len(m.ReplicaToRemove) > 0 {
-		i -= len(m.ReplicaToRemove)
-		copy(dAtA[i:], m.ReplicaToRemove)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.ReplicaToRemove)))
-		i--
 		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.ReplicaToRemove)))
+		i += copy(dAtA[i:], m.ReplicaToRemove)
 	}
-	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
-		dAtA[i] = 0x10
+	if len(m.Leader) > 0 {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
+		i += copy(dAtA[i:], m.Leader)
 	}
-	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
+	if m.LeaderEpoch != 0 {
+		dAtA[i] = 0x28
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *ExpandISROp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2639,57 +2601,48 @@ func (m *ExpandISROp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ExpandISROp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ExpandISROp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.LeaderEpoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
-		i--
-		dAtA[i] = 0x28
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
-	if len(m.Leader) > 0 {
-		i -= len(m.Leader)
-		copy(dAtA[i:], m.Leader)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
-		i--
-		dAtA[i] = 0x22
+	if m.Partition != 0 {
+		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
 	if len(m.ReplicaToAdd) > 0 {
-		i -= len(m.ReplicaToAdd)
-		copy(dAtA[i:], m.ReplicaToAdd)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.ReplicaToAdd)))
-		i--
 		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.ReplicaToAdd)))
+		i += copy(dAtA[i:], m.ReplicaToAdd)
 	}
-	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
-		dAtA[i] = 0x10
+	if len(m.Leader) > 0 {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
+		i += copy(dAtA[i:], m.Leader)
 	}
-	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
+	if m.LeaderEpoch != 0 {
+		dAtA[i] = 0x28
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *DeleteStreamOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2697,33 +2650,26 @@ func (m *DeleteStreamOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *DeleteStreamOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *DeleteStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	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
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *PauseStreamOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2731,28 +2677,15 @@ func (m *PauseStreamOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PauseStreamOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PauseStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.ResumeAll {
-		i--
-		if m.ResumeAll {
-			dAtA[i] = 1
-		} else {
-			dAtA[i] = 0
-		}
-		i--
-		dAtA[i] = 0x18
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
 	if len(m.Partitions) > 0 {
 		dAtA12 := make([]byte, len(m.Partitions)*10)
@@ -2767,26 +2700,31 @@ func (m *PauseStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 			dAtA12[j11] = uint8(num)
 			j11++
 		}
-		i -= j11
-		copy(dAtA[i:], dAtA12[:j11])
-		i = encodeVarintInternal(dAtA, i, uint64(j11))
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(j11))
+		i += copy(dAtA[i:], dAtA12[:j11])
 	}
-	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
+	if m.ResumeAll {
+		dAtA[i] = 0x18
+		i++
+		if m.ResumeAll {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i++
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *ResumeStreamOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2794,18 +2732,15 @@ func (m *ResumeStreamOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ResumeStreamOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ResumeStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
 	if len(m.Partitions) > 0 {
 		dAtA14 := make([]byte, len(m.Partitions)*10)
@@ -2820,26 +2755,21 @@ func (m *ResumeStreamOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 			dAtA14[j13] = uint8(num)
 			j13++
 		}
-		i -= j13
-		copy(dAtA[i:], dAtA14[:j13])
-		i = encodeVarintInternal(dAtA, i, uint64(j13))
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(j13))
+		i += copy(dAtA[i:], dAtA14[:j13])
 	}
-	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
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *ReportLeaderOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2847,57 +2777,48 @@ func (m *ReportLeaderOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ReportLeaderOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ReportLeaderOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.LeaderEpoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
-		i--
-		dAtA[i] = 0x28
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
-	if len(m.Leader) > 0 {
-		i -= len(m.Leader)
-		copy(dAtA[i:], m.Leader)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
-		i--
-		dAtA[i] = 0x22
+	if m.Partition != 0 {
+		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
 	if len(m.Replica) > 0 {
-		i -= len(m.Replica)
-		copy(dAtA[i:], m.Replica)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Replica)))
-		i--
 		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Replica)))
+		i += copy(dAtA[i:], m.Replica)
 	}
-	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
-		dAtA[i] = 0x10
+	if len(m.Leader) > 0 {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
+		i += copy(dAtA[i:], m.Leader)
 	}
-	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
+	if m.LeaderEpoch != 0 {
+		dAtA[i] = 0x28
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *ChangeLeaderOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2905,45 +2826,37 @@ func (m *ChangeLeaderOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ChangeLeaderOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ChangeLeaderOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if len(m.Leader) > 0 {
-		i -= len(m.Leader)
-		copy(dAtA[i:], m.Leader)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
-		i--
-		dAtA[i] = 0x1a
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
 	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
 		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
-	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
+	if len(m.Leader) > 0 {
+		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
+		i += copy(dAtA[i:], m.Leader)
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *PublishActivityOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2951,31 +2864,25 @@ func (m *PublishActivityOp) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PublishActivityOp) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PublishActivityOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.RaftIndex != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.RaftIndex))
-		i--
 		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.RaftIndex))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *SetStreamReadonlyOp) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -2983,28 +2890,15 @@ func (m *SetStreamReadonlyOp) Marshal() (dAtA []byte, err error) {
 }
 
 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)
+	var i int
 	_ = 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.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
 	if len(m.Partitions) > 0 {
 		dAtA16 := make([]byte, len(m.Partitions)*10)
@@ -3019,26 +2913,31 @@ func (m *SetStreamReadonlyOp) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 			dAtA16[j15] = uint8(num)
 			j15++
 		}
-		i -= j15
-		copy(dAtA[i:], dAtA16[:j15])
-		i = encodeVarintInternal(dAtA, i, uint64(j15))
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(j15))
+		i += copy(dAtA[i:], dAtA16[:j15])
 	}
-	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
+	if m.Readonly {
+		dAtA[i] = 0x18
+		i++
+		if m.Readonly {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i++
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *NullableInt64) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3046,31 +2945,25 @@ func (m *NullableInt64) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *NullableInt64) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *NullableInt64) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.Value != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Value))
-		i--
 		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Value))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *NullableInt32) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3078,31 +2971,25 @@ func (m *NullableInt32) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *NullableInt32) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *NullableInt32) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.Value != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Value))
-		i--
 		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Value))
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *NullableBool) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3110,36 +2997,30 @@ func (m *NullableBool) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *NullableBool) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *NullableBool) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.Value {
-		i--
+		dAtA[i] = 0x8
+		i++
 		if m.Value {
 			dAtA[i] = 1
 		} else {
 			dAtA[i] = 0
 		}
-		i--
-		dAtA[i] = 0x8
+		i++
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *StreamConfig) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3147,158 +3028,140 @@ func (m *StreamConfig) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *StreamConfig) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *StreamConfig) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.MinIsr != nil {
-		{
-			size, err := m.MinIsr.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.RetentionMaxBytes != nil {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.RetentionMaxBytes.Size()))
+		n17, err17 := m.RetentionMaxBytes.MarshalTo(dAtA[i:])
+		if err17 != nil {
+			return 0, err17
 		}
-		i--
-		dAtA[i] = 0x5a
+		i += n17
 	}
-	if m.AutoPauseDisableIfSubscribers != nil {
-		{
-			size, err := m.AutoPauseDisableIfSubscribers.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.RetentionMaxMessages != nil {
+		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.RetentionMaxMessages.Size()))
+		n18, err18 := m.RetentionMaxMessages.MarshalTo(dAtA[i:])
+		if err18 != nil {
+			return 0, err18
 		}
-		i--
-		dAtA[i] = 0x52
+		i += n18
 	}
-	if m.AutoPauseTime != nil {
-		{
-			size, err := m.AutoPauseTime.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.RetentionMaxAge != nil {
+		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.RetentionMaxAge.Size()))
+		n19, err19 := m.RetentionMaxAge.MarshalTo(dAtA[i:])
+		if err19 != nil {
+			return 0, err19
 		}
-		i--
-		dAtA[i] = 0x4a
+		i += n19
 	}
-	if m.CompactEnabled != nil {
-		{
-			size, err := m.CompactEnabled.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.CleanerInterval != nil {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.CleanerInterval.Size()))
+		n20, err20 := m.CleanerInterval.MarshalTo(dAtA[i:])
+		if err20 != nil {
+			return 0, err20
 		}
-		i--
-		dAtA[i] = 0x42
+		i += n20
 	}
-	if m.CompactMaxGoroutines != nil {
-		{
-			size, err := m.CompactMaxGoroutines.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.SegmentMaxBytes != nil {
+		dAtA[i] = 0x2a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.SegmentMaxBytes.Size()))
+		n21, err21 := m.SegmentMaxBytes.MarshalTo(dAtA[i:])
+		if err21 != nil {
+			return 0, err21
 		}
-		i--
-		dAtA[i] = 0x3a
+		i += n21
 	}
 	if m.SegmentMaxAge != nil {
-		{
-			size, err := m.SegmentMaxAge.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
-		}
-		i--
 		dAtA[i] = 0x32
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.SegmentMaxAge.Size()))
+		n22, err22 := m.SegmentMaxAge.MarshalTo(dAtA[i:])
+		if err22 != nil {
+			return 0, err22
+		}
+		i += n22
 	}
-	if m.SegmentMaxBytes != nil {
-		{
-			size, err := m.SegmentMaxBytes.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.CompactMaxGoroutines != nil {
+		dAtA[i] = 0x3a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.CompactMaxGoroutines.Size()))
+		n23, err23 := m.CompactMaxGoroutines.MarshalTo(dAtA[i:])
+		if err23 != nil {
+			return 0, err23
 		}
-		i--
-		dAtA[i] = 0x2a
+		i += n23
 	}
-	if m.CleanerInterval != nil {
-		{
-			size, err := m.CleanerInterval.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.CompactEnabled != nil {
+		dAtA[i] = 0x42
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.CompactEnabled.Size()))
+		n24, err24 := m.CompactEnabled.MarshalTo(dAtA[i:])
+		if err24 != nil {
+			return 0, err24
 		}
-		i--
-		dAtA[i] = 0x22
+		i += n24
 	}
-	if m.RetentionMaxAge != nil {
-		{
-			size, err := m.RetentionMaxAge.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.AutoPauseTime != nil {
+		dAtA[i] = 0x4a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.AutoPauseTime.Size()))
+		n25, err25 := m.AutoPauseTime.MarshalTo(dAtA[i:])
+		if err25 != nil {
+			return 0, err25
 		}
-		i--
-		dAtA[i] = 0x1a
+		i += n25
 	}
-	if m.RetentionMaxMessages != nil {
-		{
-			size, err := m.RetentionMaxMessages.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.AutoPauseDisableIfSubscribers != nil {
+		dAtA[i] = 0x52
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.AutoPauseDisableIfSubscribers.Size()))
+		n26, err26 := m.AutoPauseDisableIfSubscribers.MarshalTo(dAtA[i:])
+		if err26 != nil {
+			return 0, err26
 		}
-		i--
-		dAtA[i] = 0x12
+		i += n26
 	}
-	if m.RetentionMaxBytes != nil {
-		{
-			size, err := m.RetentionMaxBytes.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.MinIsr != nil {
+		dAtA[i] = 0x5a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.MinIsr.Size()))
+		n27, err27 := m.MinIsr.MarshalTo(dAtA[i:])
+		if err27 != nil {
+			return 0, err27
 		}
-		i--
-		dAtA[i] = 0xa
+		i += n27
 	}
-	return len(dAtA) - i, nil
+	if m.OptimisticConcurrencyControl != nil {
+		dAtA[i] = 0x62
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.OptimisticConcurrencyControl.Size()))
+		n28, err28 := m.OptimisticConcurrencyControl.MarshalTo(dAtA[i:])
+		if err28 != nil {
+			return 0, err28
+		}
+		i += n28
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *Stream) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3306,71 +3169,59 @@ func (m *Stream) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *Stream) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *Stream) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.Name) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Name)))
+		i += copy(dAtA[i:], m.Name)
 	}
-	if m.CreationTimestamp != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.CreationTimestamp))
-		i--
-		dAtA[i] = 0x28
+	if len(m.Subject) > 0 {
+		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Subject)))
+		i += copy(dAtA[i:], m.Subject)
 	}
-	if m.Config != nil {
-		{
-			size, err := m.Config.MarshalToSizedBuffer(dAtA[:i])
+	if len(m.Partitions) > 0 {
+		for _, msg := range m.Partitions {
+			dAtA[i] = 0x1a
+			i++
+			i = encodeVarintInternal(dAtA, i, uint64(msg.Size()))
+			n, err := msg.MarshalTo(dAtA[i:])
 			if err != nil {
 				return 0, err
 			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+			i += n
 		}
-		i--
-		dAtA[i] = 0x22
 	}
-	if len(m.Partitions) > 0 {
-		for iNdEx := len(m.Partitions) - 1; iNdEx >= 0; iNdEx-- {
-			{
-				size, err := m.Partitions[iNdEx].MarshalToSizedBuffer(dAtA[:i])
-				if err != nil {
-					return 0, err
-				}
-				i -= size
-				i = encodeVarintInternal(dAtA, i, uint64(size))
-			}
-			i--
-			dAtA[i] = 0x1a
+	if m.Config != nil {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Config.Size()))
+		n29, err29 := m.Config.MarshalTo(dAtA[i:])
+		if err29 != nil {
+			return 0, err29
 		}
+		i += n29
 	}
-	if len(m.Subject) > 0 {
-		i -= len(m.Subject)
-		copy(dAtA[i:], m.Subject)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Subject)))
-		i--
-		dAtA[i] = 0x12
+	if m.CreationTimestamp != 0 {
+		dAtA[i] = 0x28
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.CreationTimestamp))
 	}
-	if len(m.Name) > 0 {
-		i -= len(m.Name)
-		copy(dAtA[i:], m.Name)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Name)))
-		i--
-		dAtA[i] = 0xa
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *Partition) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3378,112 +3229,114 @@ func (m *Partition) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *Partition) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *Partition) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.Subject) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Subject)))
+		i += copy(dAtA[i:], m.Subject)
 	}
-	if m.Readonly {
-		i--
-		if m.Readonly {
-			dAtA[i] = 1
-		} else {
-			dAtA[i] = 0
-		}
-		i--
-		dAtA[i] = 0x60
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
-	if m.Paused {
-		i--
-		if m.Paused {
-			dAtA[i] = 1
-		} else {
-			dAtA[i] = 0
-		}
-		i--
-		dAtA[i] = 0x58
+	if m.Id != 0 {
+		dAtA[i] = 0x18
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Id))
 	}
-	if m.Epoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Epoch))
-		i--
-		dAtA[i] = 0x50
+	if len(m.Group) > 0 {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Group)))
+		i += copy(dAtA[i:], m.Group)
 	}
-	if m.LeaderEpoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
-		i--
-		dAtA[i] = 0x48
+	if m.ReplicationFactor != 0 {
+		dAtA[i] = 0x28
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ReplicationFactor))
 	}
-	if len(m.Isr) > 0 {
-		for iNdEx := len(m.Isr) - 1; iNdEx >= 0; iNdEx-- {
-			i -= len(m.Isr[iNdEx])
-			copy(dAtA[i:], m.Isr[iNdEx])
-			i = encodeVarintInternal(dAtA, i, uint64(len(m.Isr[iNdEx])))
-			i--
-			dAtA[i] = 0x42
+	if len(m.Replicas) > 0 {
+		for _, s := range m.Replicas {
+			dAtA[i] = 0x32
+			i++
+			l = len(s)
+			for l >= 1<<7 {
+				dAtA[i] = uint8(uint64(l)&0x7f | 0x80)
+				l >>= 7
+				i++
+			}
+			dAtA[i] = uint8(l)
+			i++
+			i += copy(dAtA[i:], s)
 		}
 	}
 	if len(m.Leader) > 0 {
-		i -= len(m.Leader)
-		copy(dAtA[i:], m.Leader)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
-		i--
 		dAtA[i] = 0x3a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Leader)))
+		i += copy(dAtA[i:], m.Leader)
 	}
-	if len(m.Replicas) > 0 {
-		for iNdEx := len(m.Replicas) - 1; iNdEx >= 0; iNdEx-- {
-			i -= len(m.Replicas[iNdEx])
-			copy(dAtA[i:], m.Replicas[iNdEx])
-			i = encodeVarintInternal(dAtA, i, uint64(len(m.Replicas[iNdEx])))
-			i--
-			dAtA[i] = 0x32
+	if len(m.Isr) > 0 {
+		for _, s := range m.Isr {
+			dAtA[i] = 0x42
+			i++
+			l = len(s)
+			for l >= 1<<7 {
+				dAtA[i] = uint8(uint64(l)&0x7f | 0x80)
+				l >>= 7
+				i++
+			}
+			dAtA[i] = uint8(l)
+			i++
+			i += copy(dAtA[i:], s)
 		}
 	}
-	if m.ReplicationFactor != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.ReplicationFactor))
-		i--
-		dAtA[i] = 0x28
+	if m.LeaderEpoch != 0 {
+		dAtA[i] = 0x48
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
 	}
-	if len(m.Group) > 0 {
-		i -= len(m.Group)
-		copy(dAtA[i:], m.Group)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Group)))
-		i--
-		dAtA[i] = 0x22
+	if m.Epoch != 0 {
+		dAtA[i] = 0x50
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Epoch))
 	}
-	if m.Id != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Id))
-		i--
-		dAtA[i] = 0x18
+	if m.Paused {
+		dAtA[i] = 0x58
+		i++
+		if m.Paused {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i++
 	}
-	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] = 0x12
+	if m.Readonly {
+		dAtA[i] = 0x60
+		i++
+		if m.Readonly {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i++
 	}
-	if len(m.Subject) > 0 {
-		i -= len(m.Subject)
-		copy(dAtA[i:], m.Subject)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Subject)))
-		i--
-		dAtA[i] = 0xa
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *RaftJoinRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3491,40 +3344,32 @@ func (m *RaftJoinRequest) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *RaftJoinRequest) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *RaftJoinRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.NodeID) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.NodeID)))
+		i += copy(dAtA[i:], m.NodeID)
 	}
 	if len(m.NodeAddr) > 0 {
-		i -= len(m.NodeAddr)
-		copy(dAtA[i:], m.NodeAddr)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.NodeAddr)))
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.NodeAddr)))
+		i += copy(dAtA[i:], m.NodeAddr)
 	}
-	if len(m.NodeID) > 0 {
-		i -= len(m.NodeID)
-		copy(dAtA[i:], m.NodeID)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.NodeID)))
-		i--
-		dAtA[i] = 0xa
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *RaftJoinResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3532,33 +3377,26 @@ func (m *RaftJoinResponse) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *RaftJoinResponse) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *RaftJoinResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if len(m.Error) > 0 {
-		i -= len(m.Error)
-		copy(dAtA[i:], m.Error)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Error)))
-		i--
 		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Error)))
+		i += copy(dAtA[i:], m.Error)
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *MetadataSnapshot) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3566,40 +3404,32 @@ func (m *MetadataSnapshot) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *MetadataSnapshot) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *MetadataSnapshot) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if len(m.Streams) > 0 {
-		for iNdEx := len(m.Streams) - 1; iNdEx >= 0; iNdEx-- {
-			{
-				size, err := m.Streams[iNdEx].MarshalToSizedBuffer(dAtA[:i])
-				if err != nil {
-					return 0, err
-				}
-				i -= size
-				i = encodeVarintInternal(dAtA, i, uint64(size))
-			}
-			i--
+		for _, msg := range m.Streams {
 			dAtA[i] = 0xa
+			i++
+			i = encodeVarintInternal(dAtA, i, uint64(msg.Size()))
+			n, err := msg.MarshalTo(dAtA[i:])
+			if err != nil {
+				return 0, err
+			}
+			i += n
 		}
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *ReplicationRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3607,43 +3437,36 @@ func (m *ReplicationRequest) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ReplicationRequest) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ReplicationRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.LeaderEpoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
-		i--
-		dAtA[i] = 0x18
+	if len(m.ReplicaID) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.ReplicaID)))
+		i += copy(dAtA[i:], m.ReplicaID)
 	}
 	if m.Offset != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Offset))
-		i--
 		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Offset))
 	}
-	if len(m.ReplicaID) > 0 {
-		i -= len(m.ReplicaID)
-		copy(dAtA[i:], m.ReplicaID)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.ReplicaID)))
-		i--
-		dAtA[i] = 0xa
+	if m.LeaderEpoch != 0 {
+		dAtA[i] = 0x18
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *LeaderEpochOffsetRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3651,31 +3474,25 @@ func (m *LeaderEpochOffsetRequest) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *LeaderEpochOffsetRequest) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *LeaderEpochOffsetRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.LeaderEpoch != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
-		i--
 		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.LeaderEpoch))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *LeaderEpochOffsetResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3683,31 +3500,25 @@ func (m *LeaderEpochOffsetResponse) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *LeaderEpochOffsetResponse) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *LeaderEpochOffsetResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if m.EndOffset != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.EndOffset))
-		i--
 		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.EndOffset))
 	}
-	return len(dAtA) - i, nil
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	return i, nil
 }
 
 func (m *PropagatedRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3715,127 +3526,105 @@ func (m *PropagatedRequest) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PropagatedRequest) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PropagatedRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if m.Op != 0 {
+		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Op))
 	}
-	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))
+	if m.CreateStreamOp != nil {
+		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.CreateStreamOp.Size()))
+		n30, err30 := m.CreateStreamOp.MarshalTo(dAtA[i:])
+		if err30 != nil {
+			return 0, err30
 		}
-		i--
-		dAtA[i] = 0x4a
+		i += n30
 	}
-	if m.ResumeStreamOp != nil {
-		{
-			size, err := m.ResumeStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ShrinkISROp != nil {
+		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ShrinkISROp.Size()))
+		n31, err31 := m.ShrinkISROp.MarshalTo(dAtA[i:])
+		if err31 != nil {
+			return 0, err31
 		}
-		i--
-		dAtA[i] = 0x42
+		i += n31
 	}
-	if m.PauseStreamOp != nil {
-		{
-			size, err := m.PauseStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ReportLeaderOp != nil {
+		dAtA[i] = 0x22
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ReportLeaderOp.Size()))
+		n32, err32 := m.ReportLeaderOp.MarshalTo(dAtA[i:])
+		if err32 != nil {
+			return 0, err32
 		}
-		i--
-		dAtA[i] = 0x3a
+		i += n32
 	}
-	if m.DeleteStreamOp != nil {
-		{
-			size, err := m.DeleteStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ExpandISROp != nil {
+		dAtA[i] = 0x2a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ExpandISROp.Size()))
+		n33, err33 := m.ExpandISROp.MarshalTo(dAtA[i:])
+		if err33 != nil {
+			return 0, err33
 		}
-		i--
-		dAtA[i] = 0x32
+		i += n33
 	}
-	if m.ExpandISROp != nil {
-		{
-			size, err := m.ExpandISROp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.DeleteStreamOp != nil {
+		dAtA[i] = 0x32
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.DeleteStreamOp.Size()))
+		n34, err34 := m.DeleteStreamOp.MarshalTo(dAtA[i:])
+		if err34 != nil {
+			return 0, err34
 		}
-		i--
-		dAtA[i] = 0x2a
+		i += n34
 	}
-	if m.ReportLeaderOp != nil {
-		{
-			size, err := m.ReportLeaderOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.PauseStreamOp != nil {
+		dAtA[i] = 0x3a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.PauseStreamOp.Size()))
+		n35, err35 := m.PauseStreamOp.MarshalTo(dAtA[i:])
+		if err35 != nil {
+			return 0, err35
 		}
-		i--
-		dAtA[i] = 0x22
+		i += n35
 	}
-	if m.ShrinkISROp != nil {
-		{
-			size, err := m.ShrinkISROp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.ResumeStreamOp != nil {
+		dAtA[i] = 0x42
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.ResumeStreamOp.Size()))
+		n36, err36 := m.ResumeStreamOp.MarshalTo(dAtA[i:])
+		if err36 != nil {
+			return 0, err36
 		}
-		i--
-		dAtA[i] = 0x1a
+		i += n36
 	}
-	if m.CreateStreamOp != nil {
-		{
-			size, err := m.CreateStreamOp.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
+	if m.SetStreamReadonlyOp != nil {
+		dAtA[i] = 0x4a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.SetStreamReadonlyOp.Size()))
+		n37, err37 := m.SetStreamReadonlyOp.MarshalTo(dAtA[i:])
+		if err37 != nil {
+			return 0, err37
 		}
-		i--
-		dAtA[i] = 0x12
+		i += n37
 	}
-	if m.Op != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Op))
-		i--
-		dAtA[i] = 0x8
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *Error) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3843,38 +3632,31 @@ func (m *Error) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *Error) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *Error) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if m.Code != 0 {
+		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Code))
 	}
 	if len(m.Msg) > 0 {
-		i -= len(m.Msg)
-		copy(dAtA[i:], m.Msg)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Msg)))
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Msg)))
+		i += copy(dAtA[i:], m.Msg)
 	}
-	if m.Code != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Code))
-		i--
-		dAtA[i] = 0x8
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *PropagatedResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3882,43 +3664,35 @@ func (m *PropagatedResponse) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PropagatedResponse) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PropagatedResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if m.Op != 0 {
+		dAtA[i] = 0x8
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Op))
 	}
 	if m.Error != nil {
-		{
-			size, err := m.Error.MarshalToSizedBuffer(dAtA[:i])
-			if err != nil {
-				return 0, err
-			}
-			i -= size
-			i = encodeVarintInternal(dAtA, i, uint64(size))
-		}
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Error.Size()))
+		n38, err38 := m.Error.MarshalTo(dAtA[i:])
+		if err38 != nil {
+			return 0, err38
+		}
+		i += n38
 	}
-	if m.Op != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Op))
-		i--
-		dAtA[i] = 0x8
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *ServerInfoRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3926,33 +3700,26 @@ func (m *ServerInfoRequest) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ServerInfoRequest) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ServerInfoRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
 	if len(m.Id) > 0 {
-		i -= len(m.Id)
-		copy(dAtA[i:], m.Id)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Id)))
-		i--
 		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Id)))
+		i += copy(dAtA[i:], m.Id)
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *ServerInfoResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -3960,45 +3727,37 @@ func (m *ServerInfoResponse) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *ServerInfoResponse) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *ServerInfoResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.Port != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Port))
-		i--
-		dAtA[i] = 0x18
+	if len(m.Id) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Id)))
+		i += copy(dAtA[i:], m.Id)
 	}
 	if len(m.Host) > 0 {
-		i -= len(m.Host)
-		copy(dAtA[i:], m.Host)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Host)))
-		i--
 		dAtA[i] = 0x12
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Host)))
+		i += copy(dAtA[i:], m.Host)
 	}
-	if len(m.Id) > 0 {
-		i -= len(m.Id)
-		copy(dAtA[i:], m.Id)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.Id)))
-		i--
-		dAtA[i] = 0xa
+	if m.Port != 0 {
+		dAtA[i] = 0x18
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Port))
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *PartitionStatusRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -4006,38 +3765,31 @@ func (m *PartitionStatusRequest) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PartitionStatusRequest) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PartitionStatusRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
 	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
 		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
-	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
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *PartitionStatusResponse) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -4045,46 +3797,40 @@ func (m *PartitionStatusResponse) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PartitionStatusResponse) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PartitionStatusResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.IsLeader {
-		i--
-		if m.IsLeader {
+	if m.Exists {
+		dAtA[i] = 0x8
+		i++
+		if m.Exists {
 			dAtA[i] = 1
 		} else {
 			dAtA[i] = 0
 		}
-		i--
-		dAtA[i] = 0x10
+		i++
 	}
-	if m.Exists {
-		i--
-		if m.Exists {
+	if m.IsLeader {
+		dAtA[i] = 0x10
+		i++
+		if m.IsLeader {
 			dAtA[i] = 1
 		} else {
 			dAtA[i] = 0
 		}
-		i--
-		dAtA[i] = 0x8
+		i++
+	}
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *PartitionNotification) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -4092,38 +3838,31 @@ func (m *PartitionNotification) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *PartitionNotification) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *PartitionNotification) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
 	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
 		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
-	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
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func (m *Cursor) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	n, err := m.MarshalTo(dAtA)
 	if err != nil {
 		return nil, err
 	}
@@ -4131,56 +3870,46 @@ func (m *Cursor) Marshal() (dAtA []byte, err error) {
 }
 
 func (m *Cursor) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *Cursor) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
+	var i int
 	_ = i
 	var l int
 	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
+	if len(m.Stream) > 0 {
+		dAtA[i] = 0xa
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.Stream)))
+		i += copy(dAtA[i:], m.Stream)
 	}
-	if m.Offset != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Offset))
-		i--
-		dAtA[i] = 0x20
+	if m.Partition != 0 {
+		dAtA[i] = 0x10
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
 	}
 	if len(m.CursorId) > 0 {
-		i -= len(m.CursorId)
-		copy(dAtA[i:], m.CursorId)
-		i = encodeVarintInternal(dAtA, i, uint64(len(m.CursorId)))
-		i--
 		dAtA[i] = 0x1a
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(len(m.CursorId)))
+		i += copy(dAtA[i:], m.CursorId)
 	}
-	if m.Partition != 0 {
-		i = encodeVarintInternal(dAtA, i, uint64(m.Partition))
-		i--
-		dAtA[i] = 0x10
+	if m.Offset != 0 {
+		dAtA[i] = 0x20
+		i++
+		i = encodeVarintInternal(dAtA, i, uint64(m.Offset))
 	}
-	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
+	if m.XXX_unrecognized != nil {
+		i += copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	return len(dAtA) - i, nil
+	return i, nil
 }
 
 func encodeVarintInternal(dAtA []byte, offset int, v uint64) int {
-	offset -= sovInternal(v)
-	base := offset
 	for v >= 1<<7 {
 		dAtA[offset] = uint8(v&0x7f | 0x80)
 		v >>= 7
 		offset++
 	}
 	dAtA[offset] = uint8(v)
-	return base
+	return offset + 1
 }
 func (m *ServerState) Size() (n int) {
 	if m == nil {
@@ -4579,6 +4308,10 @@ func (m *StreamConfig) Size() (n int) {
 		l = m.MinIsr.Size()
 		n += 1 + l + sovInternal(uint64(l))
 	}
+	if m.OptimisticConcurrencyControl != nil {
+		l = m.OptimisticConcurrencyControl.Size()
+		n += 1 + l + sovInternal(uint64(l))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -7598,6 +7331,42 @@ func (m *StreamConfig) Unmarshal(dAtA []byte) error {
 				return err
 			}
 			iNdEx = postIndex
+		case 12:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field OptimisticConcurrencyControl", 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.OptimisticConcurrencyControl == nil {
+				m.OptimisticConcurrencyControl = &NullableBool{}
+			}
+			if err := m.OptimisticConcurrencyControl.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 63dce7c..24f5b3e 100644
--- a/server/protocol/internal.proto
+++ b/server/protocol/internal.proto
@@ -115,6 +115,7 @@ message StreamConfig  {
     NullableInt64 autoPauseTime                 = 9;
     NullableBool  autoPauseDisableIfSubscribers = 10;
     NullableInt32 minIsr                        = 11;
+    NullableBool  optimisticConcurrencyControl  = 12; 
 }
 
 message Stream {


Test patch:
diff --git a/server/api_test.go b/server/api_test.go
index ff6dbcd..6458980 100644
--- a/server/api_test.go
+++ b/server/api_test.go
@@ -1233,3 +1233,486 @@ func TestFetchPartitionMetadataReadonlyTimestamps(t *testing.T) {
 	require.Equal(t, firstReadonlyTimestamp, metadata.ReadonlyTimestamps().FirstTime())
 	require.True(t, metadata.ReadonlyTimestamps().LatestTime().After(firstReadonlyTimestamp))
 }
+
+// TestPublishAsync ensures async publish with AckHandler is able to handle async error.
+func TestPublishAsync(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+	stream := "foo"
+
+	err = client.CreateStream(context.Background(), "foo", stream)
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Publish Async
+	err = client.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader())
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+}
+
+// TestPublishAsyncWithConcurrencyNoAckPolicy ensures an error is trigger in case of concurrent publishes
+// and no AckPolicy is set
+func TestPublishAsyncWithConcurrencyNoAckPolicy(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+	stream := "foo"
+
+	// Create a stream and enable Optimistic Concurrency Control
+	err = client.CreateStream(context.Background(), "foo", stream, lift.OptimisticConcurrencyControl(true))
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Publish Async with expected offset
+	err = client.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		// Set AckPolicy to NONE
+		lift.AckPolicyNone(),
+		// expected offset
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		require.Error(t, err)
+		st := status.Convert(err)
+		// Expect error
+		require.Equal(t, "stream with concurrency control must have AckPolicy set", st.Message())
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+}
+
+// TestPublishAsyncWithConcurrencyErrorWrongOffset ensures an error is trigger in case of concurrent publishes.
+func TestPublishAsyncWithConcurrencyErrorWrongOffset(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+	stream := "foo"
+
+	// Create a stream and enable Optimistic Concurrency Control
+	err = client.CreateStream(context.Background(), "foo", stream, lift.OptimisticConcurrencyControl(true))
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Publish Async with expected offset
+	err = client.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// Wrong offset
+		lift.ExpectedOffset(100),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		require.Error(t, err)
+		st := status.Convert(err)
+		// Expect error
+		require.Equal(t, "incorrect expected offset", st.Message())
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+}
+
+// TestPublishAsyncWithConcurrencyIgnoreOffset ensures that the server will ignore concurrency control
+// for request that has expected offset -1.
+func TestPublishAsyncWithConcurrencyIgnoreOffset(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+	stream := "foo"
+
+	// Create a stream and enable Optimistic Concurrency Control
+	err = client.CreateStream(context.Background(), "foo", stream, lift.OptimisticConcurrencyControl(true))
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Publish Async with expected offset
+	err = client.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// expected offset -1 to indicate next offset
+		lift.ExpectedOffset(-1),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+}
+
+// TestPublishAsyncWithConcurrencyCorrectOffset ensures an that publish messages
+// work with correct expected offset provided
+func TestPublishAsyncWithConcurrencyCorrectOffset(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	client, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client.Close()
+	stream := "foo"
+
+	// Create a stream and enable Optimistic Concurrency Control
+	err = client.CreateStream(context.Background(), "foo", stream, lift.OptimisticConcurrencyControl(true))
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Publish Async with expected offset
+	err = client.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// Correct offset is 0 (first message)
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	// Publish Async with expected offset (2nd times)
+	err = client.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// Correct offset is 1 (second message)
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+}
+
+// TestMultiplePublishAsyncWithConcurrency ensures that in the scenario
+// where multiple publishes are epxected, the concurrency control should raise error
+// on conflicts
+func TestMultiplePublishAsyncWithConcurrency(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	// Client 1
+	client1, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client1.Close()
+
+	client2, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client2.Close()
+
+	stream := "foo"
+
+	// Create a stream and enable Optimistic Concurrency Control
+	err = client1.CreateStream(context.Background(), "foo", stream, lift.OptimisticConcurrencyControl(true))
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Client 1 publish first
+	err = client1.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// expected offset is 0 for cilent 1
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error for client 1
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+
+	// Client 2 publish late
+	err = client2.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// expected offset is 0 for the client 2
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should have concurrency error for client 2
+		require.Error(t, err)
+		st := status.Convert(err)
+		// Expect error
+		require.Equal(t, "incorrect expected offset", st.Message())
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+
+	// Client 2 publish again with correct offset
+	err = client1.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// expected offset is 1 for cilent 2
+		lift.ExpectedOffset(1),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error for client 2
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+}
+
+// TestMultiplePublishAsyncWithConcurrencyRetryWithFetchMetadata ensures that in the scenario
+// where multiple publishes are epxected, the concurrency control should raise error
+// on conflicts. Client may try to fetch latest message offset using FetchPartitionMetadata endpoint and retry
+// publishing
+func TestMultiplePublishAsyncWithConcurrencyRetryWithFetchMetadata(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.EmbeddedNATS = false
+	s1 := runServerWithConfig(t, s1Config)
+	defer s1.Stop()
+
+	getMetadataLeader(t, 10*time.Second, s1)
+
+	// Client 1
+	client1, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client1.Close()
+
+	client2, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client2.Close()
+
+	stream := "foo"
+
+	// Create a stream and enable Optimistic Concurrency Control
+	err = client1.CreateStream(context.Background(), "foo", stream, lift.OptimisticConcurrencyControl(true))
+	require.NoError(t, err)
+
+	// channel for async error handler
+	errorC := make(chan error)
+
+	// Client 1 publish first
+	err = client1.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// expected offset is 0 for cilent 1
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error for client 1
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+
+	// Client 2 publish late
+	err = client2.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		// expected offset is 0 for the client 2
+		lift.ExpectedOffset(0),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should have concurrency error for client 2
+		require.Error(t, err)
+		st := status.Convert(err)
+		// Expect error
+		require.Equal(t, "incorrect expected offset", st.Message())
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+
+	// Client 2 fetches metadata to know what position available
+	metadata, err := client2.FetchPartitionMetadata(context.Background(), stream, 0)
+	require.NoError(t, err)
+
+	latestOffet := metadata.NewestOffset()
+	// In this scenario, only one message is published, so the expected offset should be 1
+	expectedOffset := latestOffet + 1
+
+	// Client 2 publish again with correct offset
+	err = client1.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		lift.ExpectedOffset(expectedOffset),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should be no error for client 2
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+
+	// A client 3 that publishes with incorrect expected offset
+	client3, err := lift.Connect([]string{"localhost:5050"})
+	require.NoError(t, err)
+	defer client3.Close()
+
+	err = client3.PublishAsync(context.Background(), "foo", []byte("hello"),
+		func(ack *lift.Ack, err error) {
+			errorC <- err
+		},
+		lift.AckPolicyLeader(),
+		lift.ExpectedOffset(100),
+	)
+	require.NoError(t, err)
+
+	select {
+	case err := <-errorC:
+		// should have concurrency error for client 3
+		require.Error(t, err)
+		st := status.Convert(err)
+		// Expect error
+		require.Equal(t, "incorrect expected offset", st.Message())
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive expected error")
+	}
+
+}
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index 713db43..82cd760 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -60,7 +60,7 @@ func TestAppendMessageSet(t *testing.T) {
 	defer l.Close()
 	defer cleanup()
 
-	set, _, err := newMessageSetFromProto(0, 0, msgs)
+	set, _, err := newMessageSetFromProto(0, 0, msgs, false)
 	require.NoError(t, err)
 
 	offsets, err := l.AppendMessageSet(set)
diff --git a/server/commitlog/delete_cleaner_test.go b/server/commitlog/delete_cleaner_test.go
index 9dc1a33..fdfcdfa 100644
--- a/server/commitlog/delete_cleaner_test.go
+++ b/server/commitlog/delete_cleaner_test.go
@@ -205,7 +205,7 @@ func TestDeleteCleanerAge(t *testing.T) {
 	for i := 0; i < 20; i++ {
 		segs[i] = createSegment(t, dir, int64(i), 20)
 		ms, entries, err := newMessageSetFromProto(int64(i), 0,
-			[]*Message{{Timestamp: int64(i * 10)}})
+			[]*Message{{Timestamp: int64(i * 10)}}, false)
 		require.NoError(t, err)
 		require.NoError(t, segs[i].WriteMessageSet(ms, entries))
 	}
@@ -238,7 +238,7 @@ func TestDeleteCleanerMessagesBelowAgeLimit(t *testing.T) {
 	for i := 0; i < 5; i++ {
 		expected[i] = createSegment(t, dir, int64(i), 20)
 		ms, entries, err := newMessageSetFromProto(int64(i), 0,
-			[]*Message{{Timestamp: int64(i * 10)}})
+			[]*Message{{Timestamp: int64(i * 10)}}, false)
 		require.NoError(t, err)
 		require.NoError(t, expected[i].WriteMessageSet(ms, entries))
 	}
@@ -309,8 +309,7 @@ func writeToSegment(t *testing.T, seg *segment, offset int64, data []byte) {
 				LeaderEpoch: 42,
 				Value:       data,
 			},
-		},
-	)
+		}, false)
 	require.NoError(t, err)
 	require.NoError(t, seg.WriteMessageSet(ms, entries))
 }
diff --git a/server/config_test.go b/server/config_test.go
index e919b90..c17df28 100644
--- a/server/config_test.go
+++ b/server/config_test.go
@@ -39,6 +39,7 @@ func TestNewConfigFromFile(t *testing.T) {
 	require.Equal(t, time.Minute, config.Streams.SegmentMaxAge)
 	require.True(t, config.Streams.Compact)
 	require.Equal(t, 2, config.Streams.CompactMaxGoroutines)
+	require.Equal(t, false, config.Streams.ConcurrencyControl)
 
 	require.Equal(t, "foo", config.Clustering.ServerID)
 	require.Equal(t, "bar", config.Clustering.Namespace)
@@ -171,6 +172,7 @@ func TestStreamsConfigApplyOverrides(t *testing.T) {
 		AutoPauseTime:                 &proto.NullableInt64{Value: 1000000},
 		AutoPauseDisableIfSubscribers: &proto.NullableBool{Value: true},
 		MinIsr:                        &proto.NullableInt32{Value: 11},
+		OptimisticConcurrencyControl:  &proto.NullableBool{Value: true},
 	}
 	streamConfig := StreamsConfig{}
 
@@ -189,6 +191,7 @@ func TestStreamsConfigApplyOverrides(t *testing.T) {
 	require.Equal(t, s, streamConfig.AutoPauseTime)
 	require.True(t, streamConfig.AutoPauseDisableIfSubscribers)
 	require.Equal(t, 11, streamConfig.MinISR)
+	require.Equal(t, true, streamConfig.ConcurrencyControl)
 }
 
 // Ensure default stream configs are always present. This should be the case
@@ -222,6 +225,7 @@ func TestStreamsConfigApplyOverridesDefault(t *testing.T) {
 	require.Equal(t, s, streamConfig.SegmentMaxAge)
 	require.Equal(t, autoPauseTime, streamConfig.AutoPauseTime)
 	require.Equal(t, 2, streamConfig.MinISR)
+	require.Equal(t, false, streamConfig.ConcurrencyControl)
 
 	// Ensure values from custom configs overwrite default configs
 	require.Equal(t, int64(1024), streamConfig.RetentionMaxBytes)
diff --git a/server/replicator_test.go b/server/replicator_test.go
index 5bf58da..387ca95 100644
--- a/server/replicator_test.go
+++ b/server/replicator_test.go
@@ -230,6 +230,7 @@ func TestCommitOnISRShrink(t *testing.T) {
 
 	// Configure second server.
 	s2Config := getTestConfig("b", false, 5051)
+	s2Config.EmbeddedNATS = false
 	s2Config.Clustering.ReplicaMaxLagTime = time.Second
 	s2Config.Clustering.ReplicaFetchTimeout = 100 * time.Millisecond
 	s2 := runServerWithConfig(t, s2Config)
@@ -237,6 +238,7 @@ func TestCommitOnISRShrink(t *testing.T) {
 
 	// Configure third server.
 	s3Config := getTestConfig("c", false, 5052)
+	s3Config.EmbeddedNATS = false
 	s3Config.Clustering.ReplicaMaxLagTime = time.Second
 	s3Config.Clustering.ReplicaFetchTimeout = 100 * time.Millisecond
 	s3 := runServerWithConfig(t, s3Config)