Skip to content

Commit

Permalink
update some data types in Kafka config to safely store their max values
Browse files Browse the repository at this point in the history
  • Loading branch information
@loosla
loosla committed Sep 17, 2024
1 parent 3738a2f commit c8e9ba3
Show file tree
Hide file tree
Showing 2 changed files with 30 additions and 28 deletions.
35 changes: 18 additions & 17 deletions databases.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package godo
import (
"context"
"fmt"
"math/big"
"net/http"
"strings"
"time"
Expand Down Expand Up @@ -663,23 +664,23 @@ type MongoDBConfig struct {

// KafkaConfig holds advanced configurations for Kafka database clusters.
type KafkaConfig struct {
GroupInitialRebalanceDelayMs *int `json:"group_initial_rebalance_delay_ms,omitempty"`
GroupMinSessionTimeoutMs *int `json:"group_min_session_timeout_ms,omitempty"`
GroupMaxSessionTimeoutMs *int `json:"group_max_session_timeout_ms,omitempty"`
MessageMaxBytes *int `json:"message_max_bytes,omitempty"`
LogCleanerDeleteRetentionMs *int `json:"log_cleaner_delete_retention_ms,omitempty"`
LogCleanerMinCompactionLagMs *int `json:"log_cleaner_min_compaction_lag_ms,omitempty"`
LogFlushIntervalMs *int `json:"log_flush_interval_ms,omitempty"`
LogIndexIntervalBytes *int `json:"log_index_interval_bytes,omitempty"`
LogMessageDownconversionEnable *bool `json:"log_message_downconversion_enable,omitempty"`
LogMessageTimestampDifferenceMaxMs *int `json:"log_message_timestamp_difference_max_ms,omitempty"`
LogPreallocate *bool `json:"log_preallocate,omitempty"`
LogRetentionBytes *int `json:"log_retention_bytes,omitempty"`
LogRetentionHours *int `json:"log_retention_hours,omitempty"`
LogRetentionMs *int `json:"log_retention_ms,omitempty"`
LogRollJitterMs *int `json:"log_roll_jitter_ms,omitempty"`
LogSegmentDeleteDelayMs *int `json:"log_segment_delete_delay_ms,omitempty"`
AutoCreateTopicsEnable *bool `json:"auto_create_topics_enable,omitempty"`
GroupInitialRebalanceDelayMs *int `json:"group_initial_rebalance_delay_ms,omitempty"`
GroupMinSessionTimeoutMs *int `json:"group_min_session_timeout_ms,omitempty"`
GroupMaxSessionTimeoutMs *int `json:"group_max_session_timeout_ms,omitempty"`
MessageMaxBytes *int `json:"message_max_bytes,omitempty"`
LogCleanerDeleteRetentionMs *int64 `json:"log_cleaner_delete_retention_ms,omitempty"`
LogCleanerMinCompactionLagMs *uint64 `json:"log_cleaner_min_compaction_lag_ms,omitempty"`
LogFlushIntervalMs *uint64 `json:"log_flush_interval_ms,omitempty"`
LogIndexIntervalBytes *int `json:"log_index_interval_bytes,omitempty"`
LogMessageDownconversionEnable *bool `json:"log_message_downconversion_enable,omitempty"`
LogMessageTimestampDifferenceMaxMs *uint64 `json:"log_message_timestamp_difference_max_ms,omitempty"`
LogPreallocate *bool `json:"log_preallocate,omitempty"`
LogRetentionBytes *big.Int `json:"log_retention_bytes,omitempty"`
LogRetentionHours *int `json:"log_retention_hours,omitempty"`
LogRetentionMs *big.Int `json:"log_retention_ms,omitempty"`
LogRollJitterMs *uint64 `json:"log_roll_jitter_ms,omitempty"`
LogSegmentDeleteDelayMs *int `json:"log_segment_delete_delay_ms,omitempty"`
AutoCreateTopicsEnable *bool `json:"auto_create_topics_enable,omitempty"`
}

type databaseUserRoot struct {
Expand Down
23 changes: 12 additions & 11 deletions databases_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package godo
import (
"encoding/json"
"fmt"
"math/big"
"net/http"
"testing"
"time"
Expand Down Expand Up @@ -3085,7 +3086,7 @@ func TestDatabases_GetConfigKafka(t *testing.T) {
"log_message_downconversion_enable": true,
"log_message_timestamp_difference_max_ms": 120000,
"log_preallocate": false,
"log_retention_bytes": 1,
"log_retention_bytes": -1,
"log_retention_hours": 168,
"log_retention_ms": 604800000,
"log_roll_jitter_ms": 0,
Expand All @@ -3099,17 +3100,17 @@ func TestDatabases_GetConfigKafka(t *testing.T) {
GroupMinSessionTimeoutMs: PtrTo(6000),
GroupMaxSessionTimeoutMs: PtrTo(1800000),
MessageMaxBytes: PtrTo(1048588),
LogCleanerDeleteRetentionMs: PtrTo(86400000),
LogCleanerMinCompactionLagMs: PtrTo(0),
LogFlushIntervalMs: PtrTo(60000),
LogCleanerDeleteRetentionMs: PtrTo(int64(86400000)),
LogCleanerMinCompactionLagMs: PtrTo(uint64(0)),
LogFlushIntervalMs: PtrTo(uint64(60000)),
LogIndexIntervalBytes: PtrTo(4096),
LogMessageDownconversionEnable: PtrTo(true),
LogMessageTimestampDifferenceMaxMs: PtrTo(120000),
LogMessageTimestampDifferenceMaxMs: PtrTo(uint64(120000)),
LogPreallocate: PtrTo(false),
LogRetentionBytes: PtrTo(1),
LogRetentionBytes: big.NewInt(int64(-1)),
LogRetentionHours: PtrTo(168),
LogRetentionMs: PtrTo(604800000),
LogRollJitterMs: PtrTo(0),
LogRetentionMs: big.NewInt(int64(604800000)),
LogRollJitterMs: PtrTo(uint64(0)),
LogSegmentDeleteDelayMs: PtrTo(60000),
AutoCreateTopicsEnable: PtrTo(true),
}
Expand Down Expand Up @@ -3137,8 +3138,8 @@ func TestDatabases_UpdateConfigKafka(t *testing.T) {
GroupMinSessionTimeoutMs: PtrTo(6000),
GroupMaxSessionTimeoutMs: PtrTo(1800000),
MessageMaxBytes: PtrTo(1048588),
LogCleanerDeleteRetentionMs: PtrTo(86400000),
LogCleanerMinCompactionLagMs: PtrTo(0),
LogCleanerDeleteRetentionMs: PtrTo(int64(86400000)),
LogCleanerMinCompactionLagMs: PtrTo(uint64(0)),
}
)

Expand All @@ -3151,7 +3152,7 @@ func TestDatabases_UpdateConfigKafka(t *testing.T) {
require.NoError(t, err)

assert.Equal(t, b.Config, kafkaConfig)
assert.Equal(t, 0, *b.Config.LogCleanerMinCompactionLagMs, "pointers to zero value should be sent")
assert.Equal(t, uint64(0), *b.Config.LogCleanerMinCompactionLagMs, "pointers to zero value should be sent")
assert.Nil(t, b.Config.LogFlushIntervalMs, "excluded value should not be sent")

w.WriteHeader(http.StatusNoContent)
Expand Down

0 comments on commit c8e9ba3

Please sign in to comment.