Repository: xataio/pgstream
Pull request #297 (2025-04-25 10:44:25)

Update kafka config
This PR updates the kafka configuration, so that the env variables for the reader and the writer servers are different, since it causes ambiguity. The default values for the consumer group id and the partitions are used instead of being required.



Patch:
diff --git a/cmd/config/config_env.go b/cmd/config/config_env.go
index 7b0e6e6..53cfabc 100644
--- a/cmd/config/config_env.go
+++ b/cmd/config/config_env.go
@@ -55,7 +55,8 @@ func init() {
 	viper.BindEnv("PGSTREAM_POSTGRES_WRITER_DISABLE_TRIGGERS")
 	viper.BindEnv("PGSTREAM_POSTGRES_WRITER_ON_CONFLICT_ACTION")
 
-	viper.BindEnv("PGSTREAM_KAFKA_SERVERS")
+	viper.BindEnv("PGSTREAM_KAFKA_READER_SERVERS")
+	viper.BindEnv("PGSTREAM_KAFKA_WRITER_SERVERS")
 	viper.BindEnv("PGSTREAM_KAFKA_TOPIC_NAME")
 	viper.BindEnv("PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID")
 	viper.BindEnv("PGSTREAM_KAFKA_READER_CONSUMER_GROUP_START_OFFSET")
@@ -200,13 +201,13 @@ func parseSchemaSnapshotConfig(pgurl string) snapshotbuilder.SchemaSnapshotConfi
 }
 
 func parseKafkaListenerConfig() *stream.KafkaListenerConfig {
-	kafkaServers := viper.GetStringSlice("PGSTREAM_KAFKA_SERVERS")
 	kafkaTopic := viper.GetString("PGSTREAM_KAFKA_TOPIC_NAME")
-	consumerGroupID := viper.GetString("PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID")
-	if len(kafkaServers) == 0 || kafkaTopic == "" || consumerGroupID == "" {
+	kafkaServers := viper.GetStringSlice("PGSTREAM_KAFKA_READER_SERVERS")
+	if len(kafkaServers) == 0 || kafkaTopic == "" {
 		return nil
 	}
 
+	consumerGroupID := viper.GetString("PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID")
 	return &stream.KafkaListenerConfig{
 		Reader:       parseKafkaReaderConfig(kafkaServers, kafkaTopic, consumerGroupID),
 		Checkpointer: parseKafkaCheckpointConfig(),
@@ -252,10 +253,9 @@ func parseProcessorConfig() (stream.ProcessorConfig, error) {
 }
 
 func parseKafkaProcessorConfig() *stream.KafkaProcessorConfig {
-	kafkaServers := viper.GetStringSlice("PGSTREAM_KAFKA_SERVERS")
 	kafkaTopic := viper.GetString("PGSTREAM_KAFKA_TOPIC_NAME")
-	topicPartitions := viper.GetInt("PGSTREAM_KAFKA_TOPIC_PARTITIONS")
-	if len(kafkaServers) == 0 || kafkaTopic == "" || topicPartitions == 0 {
+	kafkaServers := viper.GetStringSlice("PGSTREAM_KAFKA_WRITER_SERVERS")
+	if len(kafkaServers) == 0 || kafkaTopic == "" {
 		return nil
 	}
 
diff --git a/docs/README.md b/docs/README.md
index 1bb731b..fa5b18f 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -126,7 +126,7 @@ Here's a list of all the environment variables that can be used to configure the
 
 | Environment Variable                               | Default  | Required         | Description                                                                                            |
 | -------------------------------------------------- | -------- | ---------------- | ------------------------------------------------------------------------------------------------------ |
-| PGSTREAM_KAFKA_SERVERS                             | N/A      | Yes              | URLs for the Kafka servers to connect to.                                                              |
+| PGSTREAM_KAFKA_READER_SERVERS                      | N/A      | Yes              | URLs for the Kafka servers to connect to.                                                              |
 | PGSTREAM_KAFKA_TOPIC_NAME                          | N/A      | Yes              | Name of the Kafka topic to read from.                                                                  |
 | PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID            | N/A      | Yes              | Name of the Kafka consumer group for the WAL Kafka reader.                                             |
 | PGSTREAM_KAFKA_READER_CONSUMER_GROUP_START_OFFSET  | Earliest | No               | Kafka offset from which the consumer will start if there's no offset available for the consumer group. |
@@ -151,7 +151,7 @@ One of exponential/constant backoff policies can be provided for the Kafka commi
 
 | Environment Variable                    | Default | Required         | Description                                                                                         |
 | --------------------------------------- | ------- | ---------------- | --------------------------------------------------------------------------------------------------- |
-| PGSTREAM_KAFKA_SERVERS                  | N/A     | Yes              | URLs for the Kafka servers to connect to.                                                           |
+| PGSTREAM_KAFKA_WRITER_SERVERS           | N/A     | Yes              | URLs for the Kafka servers to connect to.                                                           |
 | PGSTREAM_KAFKA_TOPIC_NAME               | N/A     | Yes              | Name of the Kafka topic to write to.                                                                |
 | PGSTREAM_KAFKA_TOPIC_PARTITIONS         | 1       | No               | Number of partitions created for the Kafka topic if auto create is enabled.                         |
 | PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR | 1       | No               | Replication factor used when creating the Kafka topic if auto create is enabled.                    |
diff --git a/docs/tutorials/kafka2os_tutorial.env b/docs/tutorials/kafka2os_tutorial.env
index 73f8aff..eb63d53 100644
--- a/docs/tutorials/kafka2os_tutorial.env
+++ b/docs/tutorials/kafka2os_tutorial.env
@@ -1,5 +1,5 @@
 # Listener config
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-opensearch-consumer-group
 
diff --git a/docs/tutorials/kafka2pg_tutorial.env b/docs/tutorials/kafka2pg_tutorial.env
index c2e53a2..537e094 100644
--- a/docs/tutorials/kafka2pg_tutorial.env
+++ b/docs/tutorials/kafka2pg_tutorial.env
@@ -1,5 +1,5 @@
 # Listener config
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-postgres-consumer-group
 
diff --git a/docs/tutorials/pg2kafka_tutorial.env b/docs/tutorials/pg2kafka_tutorial.env
index 3df25f9..94bf34f 100644
--- a/docs/tutorials/pg2kafka_tutorial.env
+++ b/docs/tutorials/pg2kafka_tutorial.env
@@ -4,7 +4,7 @@ PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_WRITER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
 PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR=1
diff --git a/docs/tutorials/postgres_kafka.md b/docs/tutorials/postgres_kafka.md
index 5082e5f..331f330 100644
--- a/docs/tutorials/postgres_kafka.md
+++ b/docs/tutorials/postgres_kafka.md
@@ -156,7 +156,7 @@ We will be using a Kafka processor which will be writing the WAL events into a K
 
 ```sh
 # List of Kafka servers
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_WRITER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
 # Replication factor for the kafka topic. Defaults to 1.
@@ -205,7 +205,7 @@ PGSTREAM_POSTGRES_REPLICATION_SLOT_NAME=pgstream_tutorial_slot
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_WRITER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
 PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR=1
@@ -224,7 +224,7 @@ PGSTREAM_POSTGRES_SNAPSHOT_TABLES="*"
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost:5432?sslmode=disable"
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_WRITER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
 PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR=1
@@ -238,7 +238,7 @@ PGSTREAM_KAFKA_TOPIC_AUTO_CREATE=true
 In this case the listener will be a Kafka reader that will listen for the WAL events from the Kafka topic. The only configuration required are the kafka servers, the topic name and the consumer group id. We will use a consumer group for each of the target outputs, in this case `pgstream-postgres-consumer-group`.
 
 ```sh
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-postgres-consumer-group
 ```
@@ -273,7 +273,7 @@ The full configuration for the kafka2pg step can be put into a `kafka2pg_tutoria
 
 ```sh
 # Listener config
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-postgres-consumer-group
 
@@ -291,7 +291,7 @@ PGSTREAM_POSTGRES_WRITER_SCHEMALOG_STORE_URL="postgres://postgres:postgres@local
 The configuration for the Kafka listener is the same as for the one in the previous step, the only difference will be the name of the consumer group id, to ensure it consumes the topic independently from the postgres one.
 
 ```sh
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-opensearch-consumer-group
 ```
@@ -304,7 +304,7 @@ The full configuration for the kafka2opensearch step can be put into a `kafka2os
 
 ```sh
 # Listener config
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-opensearch-consumer-group
 
diff --git a/kafka2os.env b/kafka2os.env
index b8613ee..ac128ab 100644
--- a/kafka2os.env
+++ b/kafka2os.env
@@ -1,5 +1,5 @@
 # Listener config
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID=pgstream-consumer-group
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_START_OFFSET=earliest
diff --git a/pg2kafka.env b/pg2kafka.env
index abe5d2a..898f733 100644
--- a/pg2kafka.env
+++ b/pg2kafka.env
@@ -8,7 +8,7 @@ PGSTREAM_POSTGRES_SNAPSHOT_WORKERS=1
 
 # Processor config
 PGSTREAM_INJECTOR_STORE_POSTGRES_URL="postgres://postgres:postgres@localhost?sslmode=disable"
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_WRITER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME=pgstream
 PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
 PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR=1
diff --git a/pkg/kafka/config.go b/pkg/kafka/config.go
index 55fc7fc..b9b8441 100644
--- a/pkg/kafka/config.go
+++ b/pkg/kafka/config.go
@@ -21,9 +21,21 @@ type TopicConfig struct {
 	AutoCreate bool
 }
 
+type ReaderConfig struct {
+	Conn ConnConfig
+	// ConsumerGroupID is the ID of the consumer group to use. If not set,
+	// defaults to "pgstream-consumer-group".
+	ConsumerGroupID string
+	// ConsumerGroupStartOffset is the offset to start consuming from. If not
+	// set, defaults to "earliest".
+	ConsumerGroupStartOffset string
+}
+
 const (
-	defaultNumPartitions     = 1
-	defaultReplicationFactor = 1
+	defaultNumPartitions       = 1
+	defaultReplicationFactor   = 1
+	defaultConsumerGroupOffset = earliestOffset
+	defaultConsumerGroupID     = "pgstream-consumer-group"
 )
 
 func (c *TopicConfig) numPartitions() int {
@@ -39,3 +51,17 @@ func (c *TopicConfig) replicationFactor() int {
 	}
 	return defaultReplicationFactor
 }
+
+func (c *ReaderConfig) consumerGroupID() string {
+	if c.ConsumerGroupID != "" {
+		return c.ConsumerGroupID
+	}
+	return defaultConsumerGroupID
+}
+
+func (c *ReaderConfig) consumerGroupStartOffset() string {
+	if c.ConsumerGroupStartOffset != "" {
+		return c.ConsumerGroupStartOffset
+	}
+	return defaultConsumerGroupOffset
+}
diff --git a/pkg/kafka/kafka_reader.go b/pkg/kafka/kafka_reader.go
index efca960..9d8d05c 100644
--- a/pkg/kafka/kafka_reader.go
+++ b/pkg/kafka/kafka_reader.go
@@ -20,12 +20,6 @@ type Reader struct {
 	reader *kafka.Reader
 }
 
-type ReaderConfig struct {
-	Conn                     ConnConfig
-	ConsumerGroupID          string
-	ConsumerGroupStartOffset string
-}
-
 const (
 	earliestOffset = "earliest"
 	latestOffset   = "latest"
@@ -40,7 +34,7 @@ func NewReader(config ReaderConfig, logger loglib.Logger) (*Reader, error) {
 	})
 
 	var startOffset int64
-	switch config.ConsumerGroupStartOffset {
+	switch config.consumerGroupStartOffset() {
 	case "", earliestOffset:
 		// default to first offset
 		startOffset = kafka.FirstOffset
@@ -59,7 +53,7 @@ func NewReader(config ReaderConfig, logger loglib.Logger) (*Reader, error) {
 		reader: kafka.NewReader(kafka.ReaderConfig{
 			Brokers:        config.Conn.Servers,
 			Topic:          config.Conn.Topic.Name,
-			GroupID:        config.ConsumerGroupID,
+			GroupID:        config.consumerGroupID(),
 			MaxBytes:       maxReaderBytes, // TODO: this needs to be in sync with the broker max size
 			CommitInterval: 0,              // disabled, we call commit ourselves
 			Dialer:         dialer,


Test patch:
diff --git a/cmd/config/config_env_test.go b/cmd/config/config_env_test.go
index b61e79c..621f0e2 100644
--- a/cmd/config/config_env_test.go
+++ b/cmd/config/config_env_test.go
@@ -32,7 +32,7 @@ func Test_EnvVarsToStreamConfig(t *testing.T) {
 	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE", "1000")
 	os.Setenv("PGSTREAM_POSTGRES_SNAPSHOT_CLEAN_TARGET_DB", "true")
 
-	os.Setenv("PGSTREAM_KAFKA_SERVERS", "localhost:9092")
+	os.Setenv("PGSTREAM_KAFKA_READER_SERVERS", "localhost:9092")
 	os.Setenv("PGSTREAM_KAFKA_TOPIC_NAME", "mytopic")
 	os.Setenv("PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID", "mygroup")
 	os.Setenv("PGSTREAM_KAFKA_READER_CONSUMER_GROUP_START_OFFSET", "earliest")
@@ -53,6 +53,7 @@ func Test_EnvVarsToStreamConfig(t *testing.T) {
 	os.Setenv("PGSTREAM_POSTGRES_WRITER_DISABLE_TRIGGERS", "false")
 	os.Setenv("PGSTREAM_POSTGRES_WRITER_ON_CONFLICT_ACTION", "nothing")
 
+	os.Setenv("PGSTREAM_KAFKA_WRITER_SERVERS", "localhost:9092")
 	os.Setenv("PGSTREAM_KAFKA_TOPIC_PARTITIONS", "1")
 	os.Setenv("PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR", "1")
 	os.Setenv("PGSTREAM_KAFKA_TOPIC_AUTO_CREATE", "true")
diff --git a/cmd/config/test/test_config.env b/cmd/config/test/test_config.env
index 6257ef4..9eea900 100644
--- a/cmd/config/test/test_config.env
+++ b/cmd/config/test/test_config.env
@@ -13,7 +13,7 @@ PGSTREAM_POSTGRES_SNAPSHOT_BATCH_PAGE_SIZE=1000
 PGSTREAM_POSTGRES_SNAPSHOT_CLEAN_TARGET_DB=true
 
 # Kafka
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_READER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME="mytopic"
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_ID="mygroup"
 PGSTREAM_KAFKA_READER_CONSUMER_GROUP_START_OFFSET="earliest"
@@ -39,7 +39,7 @@ PGSTREAM_POSTGRES_WRITER_DISABLE_TRIGGERS=false
 PGSTREAM_POSTGRES_WRITER_ON_CONFLICT_ACTION="nothing"
 
 # Kafka
-PGSTREAM_KAFKA_SERVERS="localhost:9092"
+PGSTREAM_KAFKA_WRITER_SERVERS="localhost:9092"
 PGSTREAM_KAFKA_TOPIC_NAME="mytopic"
 PGSTREAM_KAFKA_TOPIC_PARTITIONS=1
 PGSTREAM_KAFKA_TOPIC_REPLICATION_FACTOR=1