Last offset fetch: prevent destination connection by returning early (#3396)

Implementation in #3367 misses
the fact that GetConnector which we call under the hood opens a
destination connection, and only after we successfully get a connector
do we assert that the connector supports T.

This PR clones GetByNameAs and passes in an `expectedType` parameter
which is used to error out with unsupported early

Author: Amogh-Bharadwaj

flow/activities/flowable_core.go

@@ -143,8 +143,8 @@ func syncCore[TPull connectors.CDCPullConnectorCore, TSync connectors.CDCSyncCon
 
 	lastOffset, err := func() (model.CdcCheckpoint, error) {
 		// special case pg-pg replication, where offsets are stored on destination instead of catalog
-		if _, isPg := any(srcConn).(*connpostgres.PostgresConnector); isPg {
-			dstPgConn, err := connectors.GetByNameAs[*connpostgres.PostgresConnector](ctx, config.Env, a.CatalogPool, config.DestinationName)
+		if _, isSourcePg := any(srcConn).(*connpostgres.PostgresConnector); isSourcePg {
+			dstPgConn, err := connectors.GetPostgresConnectorByName(ctx, config.Env, a.CatalogPool, config.DestinationName)
 			if err != nil {
 				if !errors.Is(err, errors.ErrUnsupported) {
 					return model.CdcCheckpoint{}, fmt.Errorf("failed to get destination connector to get last offset: %w", err)

flow/connectors/core.go

@@ -485,6 +485,22 @@ func GetByNameAs[T Connector](ctx context.Context, env map[string]string, catalo
 	return GetAs[T](ctx, env, peer)
 }
 
+func GetPostgresConnectorByName(
+	ctx context.Context,
+	env map[string]string,
+	catalogPool shared.CatalogPool,
+	name string,
+) (*connpostgres.PostgresConnector, error) {
+	peer, err := LoadPeer(ctx, catalogPool, name)
+	if err != nil {
+		return nil, err
+	}
+	if peer.Type != protos.DBType_POSTGRES {
+		return nil, errors.ErrUnsupported
+	}
+	return GetAs[*connpostgres.PostgresConnector](ctx, env, peer)
+}
+
 func CloseConnector(ctx context.Context, conn Connector) {
 	if err := conn.Close(); err != nil {
 		internal.LoggerFromCtx(ctx).Error("error closing connector", slog.Any("error", err))