log latest ts at eventhub

flow/connectors/eventhub/eventhub.go

@@ -133,9 +133,19 @@ func (c *EventHubConnector) processBatch(
 	batchPerTopic := NewHubBatches(c.hubManager)
 	toJSONOpts := model.NewToJSONOptions(c.config.UnnestColumns)
 
+	maxTime := time.Time{}
+
 	numRecords := 0
 	for record := range batch.GetRecords() {
 		numRecords++
+
+		tsval, err := record.GetItems().GetColValueAsTimestamp("ts")
+		if err == nil {
+			if tsval.After(maxTime) {
+				maxTime = *tsval
+			}
+		}
+
 		json, err := record.GetItems().ToJSONWithOpts(toJSONOpts)
 		if err != nil {
 			log.WithFields(log.Fields{
@@ -187,9 +197,15 @@ func (c *EventHubConnector) processBatch(
 		}
 	}
 
+	utcMaxTSString := ""
+	if !maxTime.IsZero() {
+		utcMaxTSString = maxTime.UTC().Format(time.RFC3339)
+	}
+
 	log.WithFields(log.Fields{
 		"flowName": flowJobName,
-	}).Infof("[total] successfully sent %d records to event hub", numRecords)
+	}).Infof("[total] successfully sent %d records to event hub - max ts in utc - %s",
+		numRecords, utcMaxTSString)
 	return uint32(numRecords), nil
 }
 

flow/model/model.go

@@ -45,6 +45,21 @@ type Record interface {
 	GetItems() *RecordItems
 }
 
+// get col value as timestamp (on record)
+func (r *RecordItems) GetColValueAsTimestamp(colName string) (*time.Time, error) {
+	val, err := r.GetValueByColName(colName)
+	if err != nil {
+		return nil, err
+	}
+
+	t, ok := val.Value.(time.Time)
+	if !ok {
+		return nil, fmt.Errorf("expected time.Time value for column %s", colName)
+	}
+
+	return &t, nil
+}
+
 type RecordItems struct {
 	colToValIdx map[string]int
 	values      []*qvalue.QValue