Merge branch 'main' into resync-qrep-mirror

.github/workflows/ci.yml

@@ -30,7 +30,7 @@ jobs:
           --health-timeout 5s
           --health-retries 5
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
         with:
           submodules: recursive
 

.github/workflows/customer-docker.yml

@@ -19,7 +19,7 @@ jobs:
       packages: write
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 

.github/workflows/dev-docker.yml

@@ -17,7 +17,7 @@ jobs:
       packages: write
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 

.github/workflows/flow.yml

@@ -31,11 +31,11 @@ jobs:
           --health-retries 5
     steps:
       - name: checkout sources
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - uses: actions/setup-go@v3
         with:
-          go-version: ">=1.19.0"
+          go-version: ">=1.21.0"
 
       - name: install gotestsum
         run: |

.github/workflows/golang-lint.yml

@@ -17,7 +17,7 @@ jobs:
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 

.github/workflows/rust-lint.yml

@@ -17,7 +17,7 @@ jobs:
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 

.github/workflows/stable-docker.yml

@@ -16,7 +16,7 @@ jobs:
       packages: write
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
         with:
           submodules: recursive
 

.github/workflows/ui-build.yml

@@ -17,7 +17,7 @@ jobs:
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Install Node.js dependencies
         working-directory: ui

.github/workflows/ui-lint.yml

@@ -21,7 +21,7 @@ jobs:
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Install Node.js dependencies
         working-directory: ui
@@ -34,3 +34,5 @@ jobs:
           prettier: true
           eslint_dir: ui
           prettier_dir: ui
+          eslint_args: "--max-warnings 0"
+          eslint_extensions: js,ts,jsx,tsx

flow/connectors/eventhub/eventhub.go

@@ -38,7 +38,7 @@ func NewEventHubConnector(
 		return nil, err
 	}
 
-	hubManager := NewEventHubManager(ctx, defaultAzureCreds, config)
+	hubManager := NewEventHubManager(defaultAzureCreds, config)
 	metadataSchemaName := "peerdb_eventhub_metadata" // #nosec G101
 	pgMetadata, err := metadataStore.NewPostgresMetadataStore(ctx, config.GetMetadataDb(),
 		metadataSchemaName)
@@ -124,7 +124,6 @@ func (c *EventHubConnector) updateLastOffset(jobName string, offset int64) error
 func (c *EventHubConnector) processBatch(
 	flowJobName string,
 	batch *model.CDCRecordStream,
-	eventsPerBatch int,
 	maxParallelism int64,
 ) (uint32, error) {
 	ctx := context.Background()
@@ -133,6 +132,18 @@ func (c *EventHubConnector) processBatch(
 	batchPerTopic := NewHubBatches(c.hubManager)
 	toJSONOpts := model.NewToJSONOptions(c.config.UnnestColumns)
 
+	flushBatch := func() error {
+		err := c.sendEventBatch(ctx, batchPerTopic, maxParallelism, flowJobName, tableNameRowsMapping)
+		if err != nil {
+			log.WithFields(log.Fields{
+				"flowName": flowJobName,
+			}).Infof("failed to send event batch: %v", err)
+			return err
+		}
+		batchPerTopic.Clear()
+		return nil
+	}
+
 	numRecords := 0
 	for record := range batch.GetRecords() {
 		numRecords++
@@ -144,18 +155,6 @@ func (c *EventHubConnector) processBatch(
 			return 0, err
 		}
 
-		flushBatch := func() error {
-			err := c.sendEventBatch(ctx, batchPerTopic, maxParallelism, flowJobName, tableNameRowsMapping)
-			if err != nil {
-				log.WithFields(log.Fields{
-					"flowName": flowJobName,
-				}).Infof("failed to send event batch: %v", err)
-				return err
-			}
-			batchPerTopic.Clear()
-			return nil
-		}
-
 		topicName, err := NewScopedEventhub(record.GetTableName())
 		if err != nil {
 			log.WithFields(log.Fields{
@@ -172,19 +171,20 @@ func (c *EventHubConnector) processBatch(
 			return 0, err
 		}
 
-		if (numRecords)%eventsPerBatch == 0 {
-			err := flushBatch()
-			if err != nil {
-				return 0, err
-			}
+		if numRecords%1000 == 0 {
+			log.WithFields(log.Fields{
+				"flowName": flowJobName,
+			}).Infof("processed %d records for sending", numRecords)
 		}
 	}
 
-	if batchPerTopic.Len() > 0 {
-		err := c.sendEventBatch(ctx, batchPerTopic, maxParallelism, flowJobName, tableNameRowsMapping)
-		if err != nil {
-			return 0, err
-		}
+	log.WithFields(log.Fields{
+		"flowName": flowJobName,
+	}).Infof("processed %d records for sending", numRecords)
+
+	err := flushBatch()
+	if err != nil {
+		return 0, err
 	}
 
 	log.WithFields(log.Fields{
@@ -203,10 +203,6 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 		shutdown <- true
 	}()
 
-	eventsPerBatch := int(req.PushBatchSize)
-	if eventsPerBatch <= 0 {
-		eventsPerBatch = 10000
-	}
 	maxParallelism := req.PushParallelism
 	if maxParallelism <= 0 {
 		maxParallelism = 10
@@ -221,13 +217,13 @@ func (c *EventHubConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S
 	// otherwise, we block until processBatch is done.
 	if utils.GetEnvBool("PEERDB_BETA_EVENTHUB_PUSH_ASYNC", false) {
 		go func() {
-			numRecords, err = c.processBatch(req.FlowJobName, batch, eventsPerBatch, maxParallelism)
+			numRecords, err = c.processBatch(req.FlowJobName, batch, maxParallelism)
 			if err != nil {
 				log.Errorf("[async] failed to process batch: %v", err)
 			}
 		}()
 	} else {
-		numRecords, err = c.processBatch(req.FlowJobName, batch, eventsPerBatch, maxParallelism)
+		numRecords, err = c.processBatch(req.FlowJobName, batch, maxParallelism)
 		if err != nil {
 			log.Errorf("failed to process batch: %v", err)
 			return nil, err
@@ -316,7 +312,7 @@ func (c *EventHubConnector) sendEventBatch(
 		return firstErr
 	}
 
-	log.Infof("successfully sent %d events to event hub", numEventsPushed)
+	log.Infof("[sendEventBatch] successfully sent %d events to event hub", numEventsPushed)
 	return nil
 }
 
@@ -328,7 +324,7 @@ func (c *EventHubConnector) sendBatch(
 	subCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
 	defer cancel()
 
-	hub, err := c.hubManager.GetOrCreateHubClient(tblName)
+	hub, err := c.hubManager.GetOrCreateHubClient(subCtx, tblName)
 	if err != nil {
 		return err
 	}

flow/connectors/eventhub/hubmanager.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
@@ -24,7 +25,6 @@ type EventHubManager struct {
 }
 
 func NewEventHubManager(
-	ctx context.Context,
 	creds *azidentity.DefaultAzureCredential,
 	groupConfig *protos.EventHubGroupConfig,
 ) *EventHubManager {
@@ -40,7 +40,8 @@ func NewEventHubManager(
 	}
 }
 
-func (m *EventHubManager) GetOrCreateHubClient(name ScopedEventhub) (*azeventhubs.ProducerClient, error) {
+func (m *EventHubManager) GetOrCreateHubClient(ctx context.Context, name ScopedEventhub) (
+	*azeventhubs.ProducerClient, error) {
 	ehConfig, ok := m.peerConfig.Get(name.PeerName)
 	if !ok {
 		return nil, fmt.Errorf("eventhub '%s' not configured", name)
@@ -53,9 +54,27 @@ func (m *EventHubManager) GetOrCreateHubClient(name ScopedEventhub) (*azeventhub
 		namespace = fmt.Sprintf("%s.servicebus.windows.net", namespace)
 	}
 
-	hub, ok := m.hubs.Load(name)
-	if !ok {
-		opts := &azeventhubs.ProducerClientOptions{}
+	var hubConnectOK bool
+	var hub any
+	hub, hubConnectOK = m.hubs.Load(name)
+	if hubConnectOK {
+		hubTmp := hub.(*azeventhubs.ProducerClient)
+		_, err := hubTmp.GetEventHubProperties(ctx, nil)
+		if err != nil {
+			log.Infof("eventhub %s not reachable. Will re-establish connection and re-create it. Err: %v", name, err)
+			m.hubs.Delete(name)
+			hubConnectOK = false
+		}
+	}
+
+	if !hubConnectOK {
+		opts := &azeventhubs.ProducerClientOptions{
+			RetryOptions: azeventhubs.RetryOptions{
+				MaxRetries:    32,
+				RetryDelay:    2 * time.Second,
+				MaxRetryDelay: 16 * time.Second,
+			},
+		}
 		hub, err := azeventhubs.NewProducerClient(namespace, name.Eventhub, m.creds, opts)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create eventhub client: %v", err)
@@ -67,8 +86,9 @@ func (m *EventHubManager) GetOrCreateHubClient(name ScopedEventhub) (*azeventhub
 	return hub.(*azeventhubs.ProducerClient), nil
 }
 
-func (m *EventHubManager) CreateEventDataBatch(ctx context.Context, name ScopedEventhub) (*azeventhubs.EventDataBatch, error) {
-	hub, err := m.GetOrCreateHubClient(name)
+func (m *EventHubManager) CreateEventDataBatch(ctx context.Context, name ScopedEventhub) (
+	*azeventhubs.EventDataBatch, error) {
+	hub, err := m.GetOrCreateHubClient(ctx, name)
 	if err != nil {
 		return nil, err
 	}

flow/connectors/postgres/cdc.go

@@ -191,6 +191,7 @@ func (p *PostgresCDCSource) consumeStream(
 			records.SignalAsEmpty()
 		}
 		records.RelationMessageMapping <- &p.relationMessageMapping
+		log.Infof("[finished] PullRecords streamed %d records", len(localRecords))
 	}()
 
 	shutdown := utils.HeartbeatRoutine(p.ctx, 10*time.Second, func() string {

ui/app/dto/MirrorsDTO.ts

@@ -17,4 +17,5 @@ export type TableMapRow = {
   source: string;
   destination: string;
   partitionKey: string;
+  selected: boolean;
 };

ui/app/globalstate/time.tsx

@@ -0,0 +1,13 @@
+import { create } from 'zustand';
+
+interface TZState {
+  timezone: string;
+  setZone: (tz: string) => void;
+}
+
+const useTZStore = create<TZState>()((set) => ({
+  timezone: 'UTC',
+  setZone: (tz) => set(() => ({ timezone: tz })),
+}));
+
+export default useTZStore;