Support nullabletype in custom stream config

Author: LaPetiteSouris

documentation/client_implementation.md

@@ -162,8 +162,8 @@ configure a stream. Supported options are:
 | SegmentMaxBytes | int64 |The maximum size of a single stream log segment file in bytes. Retention is always done a file at a time, so a larger segment size means fewer files but less granular control over retention. |  |
 | SegmentMaxAge | time.Duration |The maximum time before a new stream log segment is rolled out. A value of 0 means new segments will only be rolled when segment.max.bytes is reached. Retention is always done a file at a time, so a larger value means fewer files but less granular control over retention. |  |
 | CompactMaxGoroutines | int32| The maximum number of concurrent goroutines to use for compaction on a stream log (only applicable if compact.enabled is true). |  |
-| EnableCompact | | Enable message compaction by key on the server for this stream |  |
-| DisableCompact | | Disable message compaction by key on the server for this stream |  |
+| SetCompactEnabled | bool | Enable message compaction by key on the server for this stream |  |
+
 
 `CreateStream` returns/throws an error if the operation fails, specifically
 `ErrStreamExists` if a stream with the given name already exists.
@@ -894,7 +894,6 @@ thrown. Otherwise, any other error/exception is thrown if the operation failed.
 
 Also, client can set custom configurations for the stream to be created. The exhaustive list of supported stream configuration are:
 
-
 ```
 RetentionMaxBytes,
 RetentionMaxMessages,
@@ -907,7 +906,24 @@ CompactMaxGoroutines,
 ```
 
 Refer to [Sream Configuration](configuration.md#streams-configuration-settings) for more details
-Note that these opts are optional, if not given, the default configurations of the broker will be used instead.
+Note that these opts are optional, if not given, the default configurations of the broker will be used instead. 
+
+In order to differentiate between custom configuration and default server's configuration, we use 3 custom `NullableType` in setting options for the `CreateStreamRequest`. These custom types are
+
+```proto
+message NullableInt64 {
+    int64 value = 1; 
+}
+
+message NullableInt32 {
+    int32 value = 1; 
+}
+
+message NullableBool {
+    bool value = 1; 
+}
+
+```
 
 Note: if `CompactMaxGoroutines` is configured, you have to make sure manually that `CompacEnabled` is also set. The reason is that if this is not enabled explicitly, the servier will use default configuration and that may be to disable compaction on the service side, which renders `CompactMaxGoroutines` to be unused.
 
@@ -930,21 +946,14 @@ func (c *client) CreateStream(ctx context.Context, subject, name string, options
 		ReplicationFactor:    opts.ReplicationFactor,
 		Group:                opts.Group,
 		Partitions:           opts.Partitions,
-		RetentionMaxAge:      opts.RetentionMaxAge.Milliseconds(),
-		CleanerInterval:      opts.CleanerInterval.Milliseconds(),
+		RetentionMaxAge:      opts.RetentionMaxAge,
+		RetentionMaxBytes:    opts.RetentionMaxBytes,
+		RetentionMaxMessages: opts.RetentionMaxMessages,
+		CleanerInterval:      opts.CleanerInterval,
 		SegmentMaxBytes:      opts.SegmentMaxBytes,
-		SegmentMaxAge:        opts.SegmentMaxAge.Milliseconds(),
+		SegmentMaxAge:        opts.SegmentMaxAge,
 		CompactMaxGoroutines: opts.CompactMaxGoroutines,
-	}
-
-	if opts.RetentionMaxBytes != nil {
-		req.RetentionMaxBytes = opts.RetentionMaxBytes
-	}
-	if opts.RetentionMaxMessages != nil {
-		req.RetentionMaxMessages = opts.RetentionMaxMessages
-	}
-	if opts.CompactEnabled != nil {
-		req.CompactEnabled = opts.CompactEnabled
+		CompactEnabled:       opts.CompactEnabled,
 	}
 
 	err := c.doResilientRPC(func(client proto.APIClient) error {

server/api.go

@@ -56,24 +56,38 @@ func (a *apiServer) CreateStream(ctx context.Context, req *client.CreateStreamRe
 		}
 	}
 	// set custom stream config
-	streamConfig := &proto.CustomStreamConfig{
-		RetentionMaxAge:      req.RetentionMaxAge,
-		CleanerInterval:      req.CleanerInterval,
-		SegmentMaxBytes:      req.SegmentMaxBytes,
-		SegmentMaxAge:        req.SegmentMaxAge,
-		CompactMaxGoroutines: req.CompactMaxGoroutines,
+	streamConfig := &proto.CustomStreamConfig{}
+
+	if req.RetentionMaxAge != nil {
+		streamConfig.RetentionMaxAge = &proto.NullableInt64{Value: req.RetentionMaxAge.GetValue()}
+	}
+
+	if req.CleanerInterval != nil {
+		streamConfig.CleanerInterval = &proto.NullableInt64{Value: req.CleanerInterval.GetValue()}
+	}
+
+	if req.SegmentMaxBytes != nil {
+		streamConfig.SegmentMaxBytes = &proto.NullableInt64{Value: req.SegmentMaxBytes.GetValue()}
+	}
+
+	if req.SegmentMaxAge != nil {
+		streamConfig.SegmentMaxAge = &proto.NullableInt64{Value: req.SegmentMaxAge.GetValue()}
+	}
+
+	if req.CompactMaxGoroutines != nil {
+		streamConfig.CompactMaxGoroutines = &proto.NullableInt32{Value: req.CompactMaxGoroutines.GetValue()}
 	}
 
 	if req.RetentionMaxBytes != nil {
-		streamConfig.RetentionMaxBytes = &proto.RetentionMaxBytes{Value: req.RetentionMaxBytes.GetValue()}
+		streamConfig.RetentionMaxBytes = &proto.NullableInt64{Value: req.RetentionMaxBytes.GetValue()}
 	}
 
 	if req.RetentionMaxMessages != nil {
-		streamConfig.RetentionMaxMessages = &proto.RetentionMaxMessages{Value: req.RetentionMaxMessages.GetValue()}
+		streamConfig.RetentionMaxMessages = &proto.NullableInt64{Value: req.RetentionMaxMessages.GetValue()}
 	}
 
 	if req.CompactEnabled != nil {
-		streamConfig.CompactEnabled = &proto.CompactEnabled{Value: req.CompactEnabled.GetValue()}
+		streamConfig.CompactEnabled = &proto.NullableBool{Value: req.CompactEnabled.GetValue()}
 	}
 
 	stream := &proto.Stream{

server/config.go

@@ -188,16 +188,16 @@ func (l *StreamsConfig) ParseCustomStreamConfig(c *proto.CustomStreamConfig) {
 	}
 	// By default, duration configuration a considered as millisecon
 	retentionMaxAge := c.GetRetentionMaxAge()
-	if retentionMaxAge != 0 {
-		l.RetentionMaxAge = time.Duration(retentionMaxAge) * time.Millisecond
+	if retentionMaxAge != nil {
+		l.RetentionMaxAge = time.Duration(retentionMaxAge.GetValue()) * time.Millisecond
 	}
 	cleanerInterval := c.GetCleanerInterval()
-	if cleanerInterval != 0 {
-		l.CleanerInterval = time.Duration(cleanerInterval) * time.Millisecond
+	if cleanerInterval != nil {
+		l.CleanerInterval = time.Duration(cleanerInterval.GetValue()) * time.Millisecond
 	}
 	segmentMaxAge := c.GetSegmentMaxAge()
-	if segmentMaxAge != 0 {
-		l.SegmentMaxAge = time.Duration(segmentMaxAge) * time.Millisecond
+	if segmentMaxAge != nil {
+		l.SegmentMaxAge = time.Duration(segmentMaxAge.GetValue()) * time.Millisecond
 	}
 	if c.GetRetentionMaxBytes() != nil {
 		l.RetentionMaxBytes = c.GetRetentionMaxBytes().GetValue()
@@ -206,16 +206,16 @@ func (l *StreamsConfig) ParseCustomStreamConfig(c *proto.CustomStreamConfig) {
 		l.RetentionMaxMessages = c.GetRetentionMaxMessages().GetValue()
 	}
 
-	if c.GetSegmentMaxBytes() != 0 {
-		l.SegmentMaxBytes = c.GetSegmentMaxBytes()
+	if c.GetSegmentMaxBytes() != nil {
+		l.SegmentMaxBytes = c.GetSegmentMaxBytes().GetValue()
 	}
 
 	if c.GetCompactEnabled() != nil {
 		l.Compact = c.GetCompactEnabled().GetValue()
 	}
 
-	if c.GetCompactMaxGoroutines() != 0 {
-		l.CompactMaxGoroutines = int(c.GetCompactMaxGoroutines())
+	if c.GetCompactMaxGoroutines() != nil {
+		l.CompactMaxGoroutines = int(c.GetCompactMaxGoroutines().GetValue())
 	}
 
 }

server/config_test.go

@@ -131,13 +131,13 @@ func TestParseCustomStreamConfig(t *testing.T) {
 	// Given custom stream config
 	// duration configuration is in millisecond
 	customStreamConfig := &proto.CustomStreamConfig{
-		SegmentMaxBytes:      1024,
-		SegmentMaxAge:        1000000,
-		RetentionMaxBytes:    &proto.RetentionMaxBytes{Value: 2048},
-		RetentionMaxMessages: &proto.RetentionMaxMessages{Value: 1000},
-		RetentionMaxAge:      1000000,
-		CleanerInterval:      1000000,
-		CompactMaxGoroutines: 10,
+		SegmentMaxBytes:      &proto.NullableInt64{Value: 1024},
+		SegmentMaxAge:        &proto.NullableInt64{Value: 1000000},
+		RetentionMaxBytes:    &proto.NullableInt64{Value: 2048},
+		RetentionMaxMessages: &proto.NullableInt64{Value: 1000},
+		RetentionMaxAge:      &proto.NullableInt64{Value: 1000000},
+		CleanerInterval:      &proto.NullableInt64{Value: 1000000},
+		CompactMaxGoroutines: &proto.NullableInt32{Value: 10},
 	}
 	streamConfig := StreamsConfig{}
 
@@ -165,11 +165,11 @@ func TestDefaultCustomStreamConfig(t *testing.T) {
 
 	// Given custom configs
 	customStreamConfig := &proto.CustomStreamConfig{
-		RetentionMaxBytes:    &proto.RetentionMaxBytes{Value: 1024},
-		RetentionMaxMessages: &proto.RetentionMaxMessages{Value: 1000},
-		RetentionMaxAge:      1000000,
-		CleanerInterval:      1000000,
-		CompactMaxGoroutines: 10,
+		RetentionMaxBytes:    &proto.NullableInt64{Value: 1024},
+		RetentionMaxMessages: &proto.NullableInt64{Value: 1000},
+		RetentionMaxAge:      &proto.NullableInt64{Value: 1000000},
+		CleanerInterval:      &proto.NullableInt64{Value: 1000000},
+		CompactMaxGoroutines: &proto.NullableInt32{Value: 10},
 	}
 
 	streamConfig.ParseCustomStreamConfig(customStreamConfig)
@@ -195,7 +195,7 @@ func TestCompactEnabledInCustomStreamConfig(t *testing.T) {
 
 	// Given custom configs with option to disable compact
 	customStreamConfig := &proto.CustomStreamConfig{
-		CompactEnabled: &proto.CompactEnabled{Value: false},
+		CompactEnabled: &proto.NullableBool{Value: false},
 	}
 
 	streamConfig.ParseCustomStreamConfig(customStreamConfig)
@@ -207,7 +207,7 @@ func TestCompactEnabledInCustomStreamConfig(t *testing.T) {
 	streamConfig2 := StreamsConfig{}
 	// Given custom configs with option to enable compact
 	customStreamConfig2 := &proto.CustomStreamConfig{
-		CompactEnabled: &proto.CompactEnabled{Value: true},
+		CompactEnabled: &proto.NullableBool{Value: true},
 	}
 
 	streamConfig2.ParseCustomStreamConfig(customStreamConfig2)

server/partition_test.go

@@ -578,7 +578,7 @@ func TestPartitionWithCustomConfigNoError(t *testing.T) {
 	defer cleanupStorage(t)
 	server := createServer(false)
 	customStreamConfig := &proto.CustomStreamConfig{
-		RetentionMaxMessages: &proto.RetentionMaxMessages{Value: 1000},
+		RetentionMaxMessages: &proto.NullableInt64{Value: 1000},
 	}
 	p, err := server.newPartition(&proto.Partition{
 		Subject:  "foo",