feat(host/plugin): Use plugin client pkg func on job_set_sync

internal/host/plugin/job_set_sync.go

@@ -221,10 +221,6 @@ func (r *SetSyncJob) syncSets(ctx context.Context, setAggs []*hostSetAgg) error
 				setInfos: make(map[string]*setInfo),
 			}
 		}
-		ci.plg, ok = r.plugins[ag.PluginId]
-		if !ok {
-			return errors.New(ctx, errors.Internal, op, fmt.Sprintf("expected plugin %q not available", ag.PluginId))
-		}
 
 		s, err := ag.toHostSet(ctx)
 		if err != nil {
@@ -244,7 +240,8 @@ func (r *SetSyncJob) syncSets(ctx context.Context, setAggs []*hostSetAgg) error
 		catalogInfos[ag.CatalogId] = ci
 	}
 
-	// Now, look up the catalog persisted (secret) information
+	// Now, look up the catalog persisted (secret) information. Additionally,
+	// find the correct plugin to use.
 	catIds := make([]string, 0, len(catalogInfos))
 	for k := range catalogInfos {
 		catIds = append(catIds, k)
@@ -269,6 +266,11 @@ func (r *SetSyncJob) syncSets(ctx context.Context, setAggs []*hostSetAgg) error
 		ci.plgCat = plgCat
 		ci.storeCat = c
 
+		ci.plg, err = pluginClientFactoryFn(ctx, ci.plgCat, r.plugins)
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("failed to get plugin client"))
+		}
+
 		per, err := toPluginPersistedData(ctx, r.kms, c.GetProjectId(), s)
 		if err != nil {
 			return errors.Wrap(ctx, err, op)