Fix: Enforce wrangler directive config in wrangler service and pipeline transform by riyaa14 · Pull Request #1039 · data-integrations/wrangler

riyaa14 · 2026-07-14T14:06:04Z

Overview

This PR resolves issues with directive exclusions not being enforced in both the CDAP Wrangler service (wrangler-service) and the CDAP Wrangler pipeline transform (wrangler-transform).

Problem Statement

Directive exclusions and configuration rules were failing to enforce in two key areas:

CDAP Wrangler Service (wrangler-service): Directive execution and validation endpoints were not loading the system-level DirectiveConfig from backend storage (DB/store). Consequently, directive checks and validations performed via the service did not respect restriction rules.
CDAP Wrangler Transform (wrangler-transform): The pipeline transform initialized its directive parser (GrammarBasedParser) with NoOpDirectiveContext, where isExcluded() is hardcoded to return false. As a result, excluded directives were executed silently on pipeline task workers. Furthermore, in distributed execution mode, worker nodes cannot directly invoke HTTP calls to the system CDAP Wrangler service.

Solution

1. Fix for CDAP Wrangler Service

Updated service handlers (such as AbstractDirectiveHandler) to fetch DirectiveConfig from backend storage and bind it to the execution context so that UI interactions and directive evaluations via the service respect exclusions.

2. Fix for CDAP Wrangler Transform (Pipeline Runs)

Master Node Setup (prepareRun): During pipeline preparation on the master node, Wrangler.prepareRun() calls the system CDAP Wrangler service endpoint (/config) via StageContext.openConnection() to retrieve the active DirectiveConfig.
Runtime Argument Propagation: prepareRun() serializes the fetched DirectiveConfig as JSON into a CDAP pipeline runtime argument (pipeline.wrangler.directive.config).
Worker Node Execution (initialize & getRecipeParser):
- On worker initialization, getSystemDirectiveConfigFromRuntimeArgs() reads the DirectiveConfig directly from the CDAP runtime arguments, bypassing the need for worker nodes to perform HTTP calls to the service.
- Initialized GrammarBasedParser with ConfigDirectiveContext(directiveConfig) instead of NoOpDirectiveContext.
- Added eager recipe parsing during initialize() to fail fast if an excluded or restricted directive is included in the recipe before processing pipeline records.

Testing

Wrangler Interactive UI

Pipeline Preview Run

Pipeline Execution

Unit Testing

Added unit tests in WranglerDirectiveExclusionTest.
Ran existing unit test suite across wrangler-api, wrangler-service, and wrangler-transform modules.

gemini-code-assist

Code Review

This pull request implements a multi-tier configuration resolution for Wrangler directive exclusions, allowing configurations to be passed via pipeline runtime arguments to avoid unreachable HTTP service calls on workers. The feedback recommends refactoring getSystemDirectiveConfigFromRuntimeArgs to accept StageContext as a parameter and fall back to plugin properties. This refactoring completes the three-tier config resolution, enables validation during pipeline configuration, and eliminates the need for the thread-unsafe stageContext instance field.

vsethi09 · 2026-07-15T07:08:16Z

 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.api.service.http.SystemHttpServiceContext;
 import io.cdap.cdap.features.Feature;
+import io.cdap.cdap.spi.data.transaction.TransactionRunners;


SPI shouldn't be used in Handlers.

Moved TransactionRunners to ConfigStore.

vsethi09 · 2026-07-15T07:08:28Z

+        ConfigStore store = ConfigStore.get(context);
+        return store.getConfig();
+      });
+    } catch (Exception e) {


Catch specific expection

vsethi09 · 2026-07-15T07:14:16Z

+
+  protected DirectiveConfig getDirectiveConfig() {
+    try {
+      return TransactionRunners.run(getContext(), context -> {


TransactionRunner is not needed here, since ConfigStore already uses it internally.

The ConfigStore in CDAP uses TransactionRunners, but in Wrangler codebase, I didn't find any implementation of ConfigStore where TransactionRunners is being used, thus we need to add it. As suggested, I have removed TransactionRunners from Handler classes, and moved it to ConfigStore.

vsethi09 · 2026-07-15T07:24:52Z

-                                                       "precondition", "false",
-                                                       "workspaceId", workspaceId);
+      DirectiveConfig directiveConfig = getDirectiveConfig();
+      Map<String, String> properties = new HashMap<>();


Use ImmutableMap

vsethi09 · 2026-07-15T07:28:36Z

+   */
+  private DirectiveConfig fetchDirectiveConfigFromService(final StageContext context) {
+    LOG.debug("Fetching directive config from DataPrep service.");
+    HttpURLConnection connection = null;


Check if RemoteClient can be used?

RemoteClient is present in io.cdap.cdap.common.internal.remote, which is not exposed to plugins and external applications. According to CDAP Docs on Class Loading, only public API modules are available to plugins.

OpenConnection method present in cdap-api module is thus more suitable for plugin to service connection.

sahusanket · 2026-07-24T09:36:56Z

-                    .withConfigProperty(Config.NAME_DIRECTIVES);
-                }
+            DirectiveConfig directiveConfig = getSystemDirectiveConfigFromRuntimeArgs(null);
+            DirectiveContext directiveContext = new ConfigDirectiveContext(directiveConfig);


use
DirectiveContext directiveContext = new ConfigDirectiveContext(DirectiveConfig.EMPTY);

for better readability and there is no need to call getSystemDirectiveConfigFromRuntimeArgs

sahusanket · 2026-07-24T09:49:02Z

    super.prepareRun(context);

+    DirectiveConfig systemConfig = fetchDirectiveConfigFromService(context);
+    DirectiveConfig configToSet = systemConfig != null ? systemConfig : DirectiveConfig.EMPTY;


Isn't this risky?

If there are transient network failure or any unintended connectivity errors, the restricted directive will be bypassed and it may have security vulnerability .

In such cases, we should fail the pipeline.

Similar is the case with MACROs:

they retry the network call for few minutes, if it is not resolved, we fail the pipeline.

sahusanket · 2026-07-24T10:00:16Z

+   * @param context the stage context
+   * @return the fetched DirectiveConfig or null if unreachable
+   */
+  private DirectiveConfig fetchDirectiveConfigFromService(final StageContext context) {


We should be refactoring this into something similar to : AbstractServiceRetryableMacroEvaluator

Need be abstract, can be a impl class. can be DataPrepServiceClient or ServiceRetryableClient

reason :

Single responsibility ( instead of adding networking code in wrangler.java )

we need to have sufficient retry logic like exponential delay upto 5 minutes.

better exception handling like throwing IllegalStateException upon 5 minute failure..which will fail the pipeline and message should appead in pipeline logs.

intermediate exceptions in debug level for each try.

This will keep the wrangler file much clean.

please refer io.cdap.cdap.common.service.RetryStrategies for some use cases.

sahusanket · 2026-07-24T10:04:41Z


  // Configuration specifying the dataprep application and service name.
  private static final String APPLICATION_NAME = "dataprep";
  private static final String SERVICE_NAME = "service";


Move these to the new class suggest below.

sahusanket · 2026-07-24T10:08:51Z

  private static final String APPLICATION_NAME = "dataprep";
  private static final String SERVICE_NAME = "service";
  private static final String CONFIG_METHOD = "config";
+  private static final int HTTP_CONNECT_TIMEOUT_MS = 5000;


Move these to the new class suggest below. and these should be configurable

Since we do not have access to cconf , we cannot directly use it and let's support runtime arguments to handle this incase we need to modify these values in future.

private static final int DEFAULT_CONNECT_TIMEOUT_MS = 15000; // Matches cdap-default.xml http.client.connection.timeout.ms private static final int DEFAULT_READ_TIMEOUT_MS = 60000; // Matches cdap-default.xml http.client.read.timeout.ms private int getConnectTimeout(StageContext context) { String val = context.getArguments().get("wrangler.service.connect.timeout.ms"); return !Strings.isNullOrEmpty(val) ? Integer.parseInt(val) : DEFAULT_CONNECT_TIMEOUT_MS; } private int getReadTimeout(StageContext context) { String val = context.getArguments().get("wrangler.service.read.timeout.ms"); return !Strings.isNullOrEmpty(val) ? Integer.parseInt(val) : DEFAULT_READ_TIMEOUT_MS; }

sahusanket · 2026-07-24T10:17:22Z

Please add the use case of runtime argument wrangler.directive.config in exclusion-and-aliasing.md with a small example body and usecase.

vsethi09 · 2026-07-24T10:33:24Z

+    try {
+      return ConfigStore.get(getContext()).getConfig();
+    } catch (IOException | RuntimeException e) {
+      LOG.warn("Failed to fetch directive exclusions from database. Using empty configuration.", e);


If there was a failure reading from Database, IOException would be thrown. You are ignoring the exception and returning default configuration, which is incorrect.

I see, my bad, changed - throwing IOException for this method in case of error rather than handling locally.

vsethi09 · 2026-07-24T10:40:48Z

+    this.transactionRunner = null;
+  }
+
+  public ConfigStore(TransactionRunner transactionRunner) {


Is there a need for separate constructor?

This is just to make sure adding TransactionRunners is done in backward compatible way. I didn't want to change the public get method and constructor which exposed by ConfigStore currently. Thus, added new ones rather than updating the existing get method and constructor.

But rethinking this, maybe we can make this change and go forward with updating the current methods because ConfigStore is in wrangler-storage module, which is an internally used module and thus changing the constructor and methods shouldn't be a breaking change externally. If any custom plugin uses wrangler, it should be using wrangler-api module. Any insights of yours would be helpful.

vsethi09 · 2026-07-24T10:41:09Z

    }
  }

+  public static ConfigStore get(TransactionRunner transactionRunner) {


Is there a need for a separate overload?

Same reason as mentioned in #1039 (comment)

riyaa14 · 2026-07-27T08:53:00Z

Please add the use case of runtime argument wrangler.directive.config in exclusion-and-aliasing.md with a small example body and usecase.

wrangler.directive.config is strictly an internal detail, for propagating directiveConfig from master to worker nodes so that worker nodes don't have to make HTTP call to fetch it, it not a user facing runtime argument.

riyaa14 added the build Triggers unit test build label Jul 14, 2026

riyaa14 requested review from sahusanket, sidhdirenge and vsethi09 July 14, 2026 14:06

gemini-code-assist Bot reviewed Jul 14, 2026

View reviewed changes

riyaa14 force-pushed the fix/wrangler-config-api branch from 67a3374 to f33c3a9 Compare July 15, 2026 07:00

vsethi09 reviewed Jul 15, 2026

View reviewed changes

vsethi09 requested changes Jul 15, 2026

View reviewed changes

riyaa14 changed the title ~~Fix: Enforce directive exclusions in CDAP Wrangler service and pipeline transform~~ Fix: Enforce wrangler directive config in wrangler service and pipeline transform Jul 22, 2026

fix wrangler directive config API

222f064

riyaa14 force-pushed the fix/wrangler-config-api branch from f33c3a9 to 222f064 Compare July 24, 2026 06:44

sahusanket reviewed Jul 24, 2026

View reviewed changes

sahusanket requested changes Jul 24, 2026

View reviewed changes

vsethi09 reviewed Jul 24, 2026

View reviewed changes

vsethi09 requested changes Jul 24, 2026

View reviewed changes

Uh oh!

Conversation

riyaa14 commented Jul 14, 2026

Overview

Problem Statement

Solution

1. Fix for CDAP Wrangler Service

2. Fix for CDAP Wrangler Transform (Pipeline Runs)

Testing

Unit Testing

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sahusanket commented Jul 24, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

riyaa14 commented Jul 27, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants