Add --timeout parameter to stop consuming, closes #13 #CMN-122

internal/cmdparser/cmdparser.go

@@ -12,6 +12,7 @@ type CmdOptions struct {
 	Postgres string   `long:"postgres" description:"PostgreSQL connection string" env:"DBZ2PG_PGURL"`
 	Kafka    []string `long:"kafka" description:"Kafka connection string" env:"DBZ2PG_KAFKA"`
 	Topic    string   `long:"topic" description:"Topic name (or prefix of the topic name) to consume" env:"DBZ2PG_TOPIC" required:"True"`
+	Timeout  int      `long:"timeout" default:"10" description:"Idle timeout for consuming kafka messages" env:"DBZ2PG_TIMEOUT"`
 }
 
 // Parse will parse command line arguments and initialize pgengine

internal/kafka/message.go

@@ -59,10 +59,12 @@ func NewMessage(msg kafka.Message) (*Message, error) {
 	}
 	err = message.initKeys()
 	if err != nil {
+		Logger.WithError(err).Debug("initKeys failed")
 		return nil, err
 	}
 	err = message.initValues()
 	if err != nil {
+		Logger.WithError(err).Debug("initValues failed")
 		return nil, err
 	}
 	return message, nil

internal/postgres/cdc.go

@@ -6,12 +6,13 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/cybertec-postgresql/debezium2postgres/internal/kafka"
 )
 
 // Apply function reads messages from `messages` channel and applies changes to the target PostgreSQL database
-func Apply(ctx context.Context, connString string, messages <-chan kafka.Message) {
+func Apply(ctx context.Context, connString string, idleTimeout time.Duration, messages <-chan kafka.Message) {
 	conn, err := Connect(context.Background(), connString)
 	if err != nil {
 		Logger.Fatalln(err)
@@ -28,6 +29,9 @@ func Apply(ctx context.Context, connString string, messages <-chan kafka.Message
 			}
 		case <-ctx.Done():
 			return
+		case <-time.After(idleTimeout):
+			Logger.Print("Idle timeout exceeded")
+			return
 		}
 	}
 }

internal/postgres/cdc_test.go

@@ -42,7 +42,7 @@ func TestApply(t *testing.T) {
 	Connect = func(ctx context.Context, connString string) (DBExecutorContext, error) {
 		return nil, errors.New("bad connection")
 	}
-	Apply(ctx, "foo", msgChan)
+	Apply(ctx, "foo", time.Second, msgChan)
 
 	Connect = func(ctx context.Context, connString string) (DBExecutorContext, error) {
 		return &MockDbExec{
@@ -51,7 +51,7 @@ func TestApply(t *testing.T) {
 			},
 		}, nil
 	}
-	Apply(ctx, "foo", msgChan)
+	Apply(ctx, "foo", time.Second, msgChan)
 }
 
 func TestApplyCDCItem(t *testing.T) {

main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"os"
+	"time"
 
 	"github.com/cybertec-postgresql/debezium2postgres/internal/cmdparser"
 	"github.com/cybertec-postgresql/debezium2postgres/internal/kafka"
@@ -24,5 +25,5 @@ func main() {
 	// create channel for passing messages to database worker
 	var msgChannel chan kafka.Message = make(chan kafka.Message, 16)
 	kafka.Consume(context.Background(), cmdOpts.Kafka, cmdOpts.Topic, msgChannel)
-	postgres.Apply(context.Background(), cmdOpts.Postgres, msgChannel)
+	postgres.Apply(context.Background(), cmdOpts.Postgres, time.Duration(cmdOpts.Timeout)*time.Second, msgChannel)
 }