Removing Google Pub/Sub - Part 2 (#1124)

README.md

@@ -56,7 +56,6 @@ Transfer is aiming to provide coverage across all OLTPs and OLAPs databases. Cur
 
 - Message Queues
   - Kafka (default)
-  - Google Pub/Sub
 
 - [Destinations](https://artie.com/docs/destinations):
     - BigQuery

lib/config/config.go

@@ -190,7 +190,7 @@ func (c Config) Validate() error {
 			return fmt.Errorf("kafka config is nil")
 		}
 
-		// Username and password are not required (if it's within the same VPC or connecting locally
+		// Username and password may not be required if this is connecting to a private Kafka cluster.
 		if stringutil.Empty(c.Kafka.GroupID, c.Kafka.BootstrapServer) {
 			return fmt.Errorf("kafka group or bootstrap server is empty")
 		}

lib/optimization/table_data.go

@@ -51,8 +51,7 @@ type TableData struct {
 	topicConfig kafkalib.TopicConfig
 	// Partition to the latest offset(s).
 	// For Kafka, we only need the last message to commit the offset
-	// However, pub/sub requires every single message to be acked
-	PartitionsToLastMessage map[string][]artie.Message
+	PartitionsToLastMessage map[string]artie.Message
 
 	// This is used for the automatic schema detection
 	LatestCDCTs time.Time
@@ -151,7 +150,7 @@ func NewTableData(inMemoryColumns *columns.Columns, mode config.Mode, primaryKey
 		topicConfig:     topicConfig,
 		// temporaryTableSuffix is being set in `ResetTempTableSuffix`
 		temporaryTableSuffix:    "",
-		PartitionsToLastMessage: map[string][]artie.Message{},
+		PartitionsToLastMessage: map[string]artie.Message{},
 		name:                    name,
 	}
 

models/event/event.go

@@ -267,7 +267,7 @@ func (e *Event) Save(cfg config.Config, inMemDB *models.DatabaseData, tc kafkali
 	td.InsertRow(e.PrimaryKeyValue(), e.Data, e.Deleted)
 	// If the message is Kafka, then we only need the latest one
 	if message.Kind() == artie.Kafka {
-		td.PartitionsToLastMessage[message.Partition()] = []artie.Message{message}
+		td.PartitionsToLastMessage[message.Partition()] = message
 	}
 
 	td.LatestCDCTs = e.executionTime

processes/consumer/configs.go

@@ -39,16 +39,14 @@ type TopicConfigFormatter struct {
 	cdc.Format
 }
 
-func commitOffset(ctx context.Context, topic string, partitionsToOffset map[string][]artie.Message) error {
-	for _, msgs := range partitionsToOffset {
-		for _, msg := range msgs {
-			if msg.KafkaMsg != nil {
-				if err := topicToConsumer.Get(topic).CommitMessages(ctx, *msg.KafkaMsg); err != nil {
-					return err
-				}
-
-				slog.Info("Successfully committed Kafka offset", slog.String("topic", topic), slog.Int("partition", msg.KafkaMsg.Partition), slog.Int64("offset", msg.KafkaMsg.Offset))
+func commitOffset(ctx context.Context, topic string, partitionsToOffset map[string]artie.Message) error {
+	for _, msg := range partitionsToOffset {
+		if msg.KafkaMsg != nil {
+			if err := topicToConsumer.Get(topic).CommitMessages(ctx, *msg.KafkaMsg); err != nil {
+				return err
 			}
+
+			slog.Info("Successfully committed Kafka offset", slog.String("topic", topic), slog.Int("partition", msg.KafkaMsg.Partition), slog.Int64("offset", msg.KafkaMsg.Offset))
 		}
 	}