CDAP Features

CDAP Features is a Java library that provides feature flag management functionality for the CDAP (Cask Data Application Platform) ecosystem. It enables runtime control of platform features based on version compatibility and administrative configuration.

Package Information

• Package Name: cdap-features
• Package Type: Maven
• Group ID: io.cdap.cdap
• Artifact ID: cdap-features
• Language: Java
• Installation: Add Maven dependency:

  <dependency>
    <groupId>io.cdap.cdap</groupId>
    <artifactId>cdap-features</artifactId>
    <version>6.11.0</version>
  </dependency>

Core Imports

import io.cdap.cdap.features.Feature;
import io.cdap.cdap.api.feature.FeatureFlagsProvider;
import io.cdap.cdap.api.PlatformInfo;
import javax.annotation.Nullable;

Basic Usage

import io.cdap.cdap.features.Feature;
import io.cdap.cdap.api.feature.FeatureFlagsProvider;

// Assuming you have a FeatureFlagsProvider implementation
FeatureFlagsProvider provider = // ... obtained from configuration

// Check if a specific feature is enabled
boolean isEnabled = Feature.EVENT_PUBLISH.isEnabled(provider);

if (isEnabled) {
    // Use the feature
    enableEventPublishing();
}

// Get the feature flag string identifier
String flagName = Feature.EVENT_PUBLISH.getFeatureFlagString();
// Returns: "event.publish.enabled"

Architecture

The CDAP Features library is built around a simple but powerful design:

• Feature Enum: Defines all available feature flags as enumerated constants with version information
• Version-based Logic: Features are automatically enabled/disabled based on platform version compatibility
• Configuration Override: Features can be explicitly enabled/disabled through FeatureFlagsProvider configuration
• Default Behavior Control: Each feature defines its default behavior after introduction (enabled/disabled)

Capabilities

Feature Flag Management

The core functionality revolves around the Feature enum which provides version-aware feature flag management.

public enum Feature {
    REPLICATION_TRANSFORMATIONS("6.6.0"),
    EVENT_PUBLISH("6.7.0", false),
    EVENT_READER("6.10.0", false),
    PIPELINE_COMPOSITE_TRIGGERS("6.8.0"),
    PUSHDOWN_TRANSFORMATION_GROUPBY("6.7.0"),
    PUSHDOWN_TRANSFORMATION_DEDUPLICATE("6.7.0"),
    STREAMING_PIPELINE_CHECKPOINT_DELETION("6.7.1"),
    LIFECYCLE_MANAGEMENT_EDIT("6.8.0"),
    WRANGLER_FAIL_PIPELINE_FOR_ERROR("6.8.0"),
    STREAMING_PIPELINE_NATIVE_STATE_TRACKING("6.8.0", false),
    PUSHDOWN_TRANSFORMATION_WINDOWAGGREGATION("6.9.1"),
    SOURCE_CONTROL_MANAGEMENT_GIT("6.9.0"),
    SOURCE_CONTROL_MANAGEMENT_MULTI_APP("6.10.0"),
    WRANGLER_PRECONDITION_SQL("6.9.1"),
    WRANGLER_EXECUTION_SQL("6.10.0"),
    WRANGLER_SCHEMA_MANAGEMENT("6.10.0"),
    NAMESPACED_SERVICE_ACCOUNTS("6.10.0"),
    WRANGLER_KRYO_SERIALIZATION("6.10.1"),
    SOURCE_CONTROL_MANAGEMENT_GITLAB_BITBUCKET("6.10.1"),
    DATAPLANE_AUDIT_LOGGING("6.10.1");
    
    // Constructors
    Feature(String versionIntroduced);
    Feature(String versionIntroduced, boolean defaultAfterIntroduction);
    
    // Methods
    public boolean isEnabled(FeatureFlagsProvider featureFlagsProvider);
    public String getFeatureFlagString();
}

Feature Constants

Each feature constant represents a specific CDAP platform capability:

• REPLICATION_TRANSFORMATIONS: Enables data replication transformations (introduced in 6.6.0)
• EVENT_PUBLISH: Enables event publishing functionality (introduced in 6.7.0, defaults to disabled)
• EVENT_READER: Enables event reader functionality (introduced in 6.10.0, defaults to disabled)
• PIPELINE_COMPOSITE_TRIGGERS: Enables composite triggers in pipelines (introduced in 6.8.0)
• PUSHDOWN_TRANSFORMATION_GROUPBY: Enables pushdown GroupBy transformations (introduced in 6.7.0)
• PUSHDOWN_TRANSFORMATION_DEDUPLICATE: Enables pushdown deduplication transformations (introduced in 6.7.0)
• STREAMING_PIPELINE_CHECKPOINT_DELETION: Enables checkpoint deletion in streaming pipelines (introduced in 6.7.1)
• LIFECYCLE_MANAGEMENT_EDIT: Enables lifecycle management editing (introduced in 6.8.0)
• WRANGLER_FAIL_PIPELINE_FOR_ERROR: Enables pipeline failure on Wrangler errors (introduced in 6.8.0)
• STREAMING_PIPELINE_NATIVE_STATE_TRACKING: Enables native state tracking in streaming pipelines (introduced in 6.8.0, defaults to disabled)
• PUSHDOWN_TRANSFORMATION_WINDOWAGGREGATION: Enables pushdown window aggregation transformations (introduced in 6.9.1)
• SOURCE_CONTROL_MANAGEMENT_GIT: Enables Git-based source control management (introduced in 6.9.0)
• SOURCE_CONTROL_MANAGEMENT_MULTI_APP: Enables multi-application source control management (introduced in 6.10.0)
• WRANGLER_PRECONDITION_SQL: Enables SQL preconditions in Wrangler (introduced in 6.9.1)
• WRANGLER_EXECUTION_SQL: Enables SQL execution in Wrangler (introduced in 6.10.0)
• WRANGLER_SCHEMA_MANAGEMENT: Enables schema management in Wrangler (introduced in 6.10.0)
• NAMESPACED_SERVICE_ACCOUNTS: Enables namespaced service accounts (introduced in 6.10.0)
• WRANGLER_KRYO_SERIALIZATION: Enables Kryo serialization in Wrangler (introduced in 6.10.1)
• SOURCE_CONTROL_MANAGEMENT_GITLAB_BITBUCKET: Enables GitLab and Bitbucket source control management (introduced in 6.10.1)
• DATAPLANE_AUDIT_LOGGING: Enables dataplane audit logging (introduced in 6.10.1)

Feature Enablement Logic

public boolean isEnabled(FeatureFlagsProvider featureFlagsProvider)

Determines if a feature flag is enabled using the following logic:

1. First checks the featureFlagsProvider for explicit configuration
2. If not configured, falls back to version-based default behavior:
   • If current platform version ≤ feature introduction version: returns false
   • If current platform version > feature introduction version: returns the feature's default behavior (true for most features, false for some)

Parameters:

• featureFlagsProvider (FeatureFlagsProvider): Provider that contains feature flag configuration

Returns: boolean - true if the feature is enabled, false otherwise

Exception Handling: Handles NullPointerException internally and falls back to default version-based logic

Feature Flag String Identifier

public String getFeatureFlagString()

Returns the string identifier used to configure this feature flag in the provider.

Returns: String - The feature flag identifier (feature name in lowercase with underscores replaced by dots, plus ".enabled" suffix)

Example: Feature.EVENT_PUBLISH.getFeatureFlagString() returns "event.publish.enabled"

Types

FeatureFlagsProvider Interface

public interface FeatureFlagsProvider {
    /**
     * This method tells is feature flag is currently enabled. To ensure proper feature life cycle
     * management and backwards compatibility please define features using Feature enum and use 
     * Feature#isEnabled method instead of directly calling this one.
     * 
     * @param name the feature flag string identifier
     * @return value of the feature flag if set
     * @throws NullPointerException if feature flag name is not defined
     * @throws IllegalArgumentException if the feature flag is not a valid boolean
     */
    default boolean isFeatureEnabled(String name) throws NullPointerException, IllegalArgumentException;
}

Interface for providing feature flag configuration. The default implementation throws UnsupportedOperationException. Implementations should:

• Return true if the specified feature flag is enabled
• Return false if the specified feature flag is disabled
• Throw NullPointerException if the feature flag is not configured (allowing fallback to default logic)
• Throw IllegalArgumentException if the feature flag value is not a valid boolean

VersionInfo Interface

public interface VersionInfo extends Comparable<Object> {
    // Marker interface for version information that is comparable
    // All implementations must return proper version string from toString()
}

Marker interface for version information that is comparable to other instances or objects. Any implementation must return a properly formatted version string from its toString() method to enable comparison with string representations.

PlatformInfo.Version

public static final class PlatformInfo.Version implements VersionInfo {
    // Constructors
    public Version(int major, int minor, int fix, boolean snapshot, long buildTime);
    public Version(@Nullable String version);
    
    // Accessor methods
    public int getMajor();
    public int getMinor(); 
    public int getFix();
    public boolean isSnapshot();
    public long getBuildTime();
    
    // Object methods
    public String toString();
    public int compareTo(Object other);
    public boolean equals(Object obj);
    public int hashCode();
}

Version class used for comparing platform versions against feature introduction versions. This class is provided by the cdap-api dependency and implements comparison logic based on major.minor.fix version numbers.

Usage Examples

Checking Multiple Features

import io.cdap.cdap.features.Feature;
import io.cdap.cdap.api.feature.FeatureFlagsProvider;

public class FeatureManager {
    private final FeatureFlagsProvider provider;
    
    public FeatureManager(FeatureFlagsProvider provider) {
        this.provider = provider;
    }
    
    public boolean canUseAdvancedTransformations() {
        return Feature.PUSHDOWN_TRANSFORMATION_GROUPBY.isEnabled(provider) &&
               Feature.PUSHDOWN_TRANSFORMATION_DEDUPLICATE.isEnabled(provider) &&
               Feature.PUSHDOWN_TRANSFORMATION_WINDOWAGGREGATION.isEnabled(provider);
    }
    
    public boolean isSourceControlEnabled() {
        return Feature.SOURCE_CONTROL_MANAGEMENT_GIT.isEnabled(provider) ||
               Feature.SOURCE_CONTROL_MANAGEMENT_GITLAB_BITBUCKET.isEnabled(provider);
    }
}

Feature Flag Configuration

// Example of retrieving feature flag strings for configuration
import io.cdap.cdap.features.Feature;

public class ConfigurationHelper {
    public static void printAllFeatureFlags() {
        for (Feature feature : Feature.values()) {
            System.out.println(feature.name() + " -> " + feature.getFeatureFlagString());
        }
    }
}

Version-based Feature Logic

import io.cdap.cdap.features.Feature;

public class PipelineProcessor {
    private final FeatureFlagsProvider provider;
    
    public void processData() {
        // Use new checkpoint deletion if available
        if (Feature.STREAMING_PIPELINE_CHECKPOINT_DELETION.isEnabled(provider)) {
            useAdvancedCheckpointDeletion();
        } else {
            useLegacyCheckpointHandling();
        }
        
        // Enable native state tracking only if explicitly configured
        // (defaults to disabled even after version 6.8.0)
        if (Feature.STREAMING_PIPELINE_NATIVE_STATE_TRACKING.isEnabled(provider)) {
            enableNativeStateTracking();
        }
    }
}