kafka + lua scripting + flatbuffers

.github/workflows/flow.yml

@@ -86,6 +86,11 @@ jobs:
         env:
           PGPASSWORD: postgres
 
+      - name: start redpanda
+        uses: redpanda-data/[email protected]
+        with:
+          version: "latest"
+
       - name: Install Temporal CLI
         uses: temporalio/setup-temporal@v0
 

flow/activities/flowable.go

@@ -400,6 +400,7 @@ func (a *FlowableActivity) SyncFlow(
 			FlowJobName:   flowName,
 			TableMappings: options.TableMappings,
 			StagingPath:   config.CdcStagingPath,
+			Script:        config.Script,
 		})
 		if err != nil {
 			a.Alerter.LogFlowError(ctx, flowName, err)

flow/cmd/handler.go

@@ -533,9 +533,7 @@ func (h *FlowRequestHandler) CreatePeer(
 			return wrongConfigResponse, nil
 		}
 		pgConfig := pgConfigObject.PostgresConfig
-
 		encodedConfig, encodingErr = proto.Marshal(pgConfig)
-
 	case protos.DBType_SNOWFLAKE:
 		sfConfigObject, ok := config.(*protos.Peer_SnowflakeConfig)
 		if !ok {
@@ -566,13 +564,18 @@ func (h *FlowRequestHandler) CreatePeer(
 		encodedConfig, encodingErr = proto.Marshal(s3Config)
 	case protos.DBType_CLICKHOUSE:
 		chConfigObject, ok := config.(*protos.Peer_ClickhouseConfig)
-
 		if !ok {
 			return wrongConfigResponse, nil
 		}
-
 		chConfig := chConfigObject.ClickhouseConfig
 		encodedConfig, encodingErr = proto.Marshal(chConfig)
+	case protos.DBType_KAFKA:
+		kaConfigObject, ok := config.(*protos.Peer_KafkaConfig)
+		if !ok {
+			return wrongConfigResponse, nil
+		}
+		kaConfig := kaConfigObject.KafkaConfig
+		encodedConfig, encodingErr = proto.Marshal(kaConfig)
 	default:
 		return wrongConfigResponse, nil
 	}

flow/connectors/core.go

@@ -11,6 +11,7 @@ import (
 	connbigquery "github.com/PeerDB-io/peer-flow/connectors/bigquery"
 	connclickhouse "github.com/PeerDB-io/peer-flow/connectors/clickhouse"
 	conneventhub "github.com/PeerDB-io/peer-flow/connectors/eventhub"
+	connkafka "github.com/PeerDB-io/peer-flow/connectors/kafka"
 	connpostgres "github.com/PeerDB-io/peer-flow/connectors/postgres"
 	conns3 "github.com/PeerDB-io/peer-flow/connectors/s3"
 	connsnowflake "github.com/PeerDB-io/peer-flow/connectors/snowflake"
@@ -202,6 +203,8 @@ func GetConnector(ctx context.Context, config *protos.Peer) (Connector, error) {
 		return connsqlserver.NewSQLServerConnector(ctx, inner.SqlserverConfig)
 	case *protos.Peer_ClickhouseConfig:
 		return connclickhouse.NewClickhouseConnector(ctx, inner.ClickhouseConfig)
+	case *protos.Peer_KafkaConfig:
+		return connkafka.NewKafkaConnector(ctx, inner.KafkaConfig)
 	default:
 		return nil, ErrUnsupportedFunctionality
 	}
@@ -260,6 +263,7 @@ var (
 	_ CDCSyncConnector = &connbigquery.BigQueryConnector{}
 	_ CDCSyncConnector = &connsnowflake.SnowflakeConnector{}
 	_ CDCSyncConnector = &conneventhub.EventHubConnector{}
+	_ CDCSyncConnector = &connkafka.KafkaConnector{}
 	_ CDCSyncConnector = &conns3.S3Connector{}
 	_ CDCSyncConnector = &connclickhouse.ClickhouseConnector{}
 

flow/connectors/eventhub/eventhub.go

@@ -88,13 +88,7 @@ func (c *EventHubConnector) GetLastOffset(ctx context.Context, jobName string) (
 }
 
 func (c *EventHubConnector) SetLastOffset(ctx context.Context, jobName string, offset int64) error {
-	err := c.pgMetadata.UpdateLastOffset(ctx, jobName, offset)
-	if err != nil {
-		c.logger.Error("failed to update last offset", slog.Any("error", err))
-		return err
-	}
-
-	return nil
+	return c.pgMetadata.UpdateLastOffset(ctx, jobName, offset)
 }
 
 // returns the number of records synced
@@ -204,9 +198,7 @@ func (c *EventHubConnector) processBatch(
 }
 
 func (c *EventHubConnector) SyncRecords(ctx context.Context, req *model.SyncRecordsRequest) (*model.SyncResponse, error) {
-	batch := req.Records
-
-	numRecords, err := c.processBatch(ctx, req.FlowJobName, batch)
+	numRecords, err := c.processBatch(ctx, req.FlowJobName, req.Records)
 	if err != nil {
 		c.logger.Error("failed to process batch", slog.Any("error", err))
 		return nil, err

flow/connectors/external_metadata/store.go

@@ -55,6 +55,13 @@ func (p *PostgresMetadataStore) Ping(ctx context.Context) error {
 	return nil
 }
 
+func (p *PostgresMetadataStore) LogFlowInfo(ctx context.Context, flowName string, info string) error {
+	_, err := p.pool.Exec(ctx,
+		"INSERT INTO peerdb_stats.flow_errors(flow_name,error_message,error_type) VALUES($1,$2,$3)",
+		flowName, info, "info")
+	return err
+}
+
 func (p *PostgresMetadataStore) FetchLastOffset(ctx context.Context, jobName string) (int64, error) {
 	row := p.pool.QueryRow(ctx,
 		`SELECT last_offset FROM `+