apache/shardingsphere · error · PipelineInvalidParameterException

Unknown stream channel type `%s`.

Error message

Unknown stream channel type `%s`.

What it means

ALTER TRANSMISSION RULE validates the stream channel algorithm segment before persisting process configuration. It loads PipelineChannelCreator via TypedSPILoader by the configured type string; if no SPI implementation is registered under that name, findService returns empty and a PipelineInvalidParameterException is thrown with the offending type name.

Source

Thrown at kernel/data-pipeline/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/distsql/handler/transmission/update/AlterTransmissionRuleExecutor.java:50

import org.apache.shardingsphere.distsql.segment.TransmissionRuleSegment;
import org.apache.shardingsphere.infra.algorithm.core.config.AlgorithmConfiguration;
import org.apache.shardingsphere.infra.instance.metadata.InstanceType;
import org.apache.shardingsphere.infra.spi.type.typed.TypedSPILoader;
import org.apache.shardingsphere.mode.manager.ContextManager;

/**
 * Alter transmission rule executor.
 */
public final class AlterTransmissionRuleExecutor implements DistSQLUpdateExecutor<AlterTransmissionRuleStatement> {
    
    private final PipelineProcessConfigurationPersistService processConfigPersistService = new PipelineProcessConfigurationPersistService();
    
    @Override
    public void executeUpdate(final AlterTransmissionRuleStatement sqlStatement, final ContextManager contextManager) {
        PipelineProcessConfiguration processConfig = convertToProcessConfiguration(sqlStatement.getProcessConfigSegment());
        AlgorithmConfiguration streamChannel = processConfig.getStreamChannel();
        if (null != streamChannel && !TypedSPILoader.findService(PipelineChannelCreator.class, streamChannel.getType()).isPresent()) {
            throw new PipelineInvalidParameterException("Unknown stream channel type `" + streamChannel.getType() + "`.");
        }
        String jobType = TypedSPILoader.getService(PipelineJobType.class, sqlStatement.getJobTypeName()).getType();
        processConfigPersistService.persist(new PipelineContextKey(InstanceType.PROXY), jobType, processConfig);
    }
    
    private PipelineProcessConfiguration convertToProcessConfiguration(final TransmissionRuleSegment segment) {
        return new PipelineProcessConfiguration(
                convertToReadConfiguration(segment.getReadSegment()), convertToWriteConfiguration(segment.getWriteSegment()), convertToAlgorithm(segment.getStreamChannel()));
    }
    
    private PipelineReadConfiguration convertToReadConfiguration(final ReadOrWriteSegment readSegment) {
        return null == readSegment
                ? null
                : new PipelineReadConfiguration(readSegment.getWorkerThread(), readSegment.getBatchSize(), readSegment.getShardingSize(), convertToAlgorithm(readSegment.getRateLimiter()));
    }
    
    private PipelineWriteConfiguration convertToWriteConfiguration(final ReadOrWriteSegment writeSegment) {
        return null == writeSegment ? null : new PipelineWriteConfiguration(writeSegment.getWorkerThread(), writeSegment.getBatchSize(), convertToAlgorithm(writeSegment.getRateLimiter()));

View on GitHub (pinned to e952770a21)

Solutions

1. Use a built-in stream channel type shipped with the distribution (e.g. MEMORY); check available types via SHOW TRANSMISSION RULE / documentation for your version.
2. Fix typos in the STREAM CHANNEL type string.
3. For custom channels: implement PipelineChannelCreator, register it in the SPI metadata file, and place the jar in the proxy's lib directory before restarting.
4. Ensure the JDBC driver required by the channel exists in the proxy lib directory.

Example fix

-- before
ALTER TRANSMISSION RULE ... STREAM CHANNEL(NAME='mymemory');
-- after
ALTER TRANSMISSION RULE ... STREAM CHANNEL(NAME='memory');

// Before ALTER TRANSMISSION RULE, confirm the channel type is resolvable
boolean known = TypedSPILoader.findService(PipelineChannelCreator.class, "memory").isPresent();
if (!known) { throw new IllegalArgumentException("stream channel type not available"); }

try {
    proxy.execute("ALTER TRANSMISSION RULE ...");
} catch (final PipelineInvalidParameterException ex) {
    // check message for unknown type, then SHOW TRANSMISSION RULE to list valid config
}

When it happens

Trigger: Executing ALTER TRANSMATION RULE ... with a STREAM CHANNEL clause whose type (e.g. 'mymemory', 'netty', custom name) is not a registered PipelineChannelCreator SPI. Also when the JDBC driver class needed by a channel implementation is missing from the proxy classpath so the SPI fails to load.

Common situations: Typos in the channel type; writing a custom stream channel SPI and forgetting to register it in META-INF/services (or the new ServiceLoader metadata) inside the proxy lib directory; copying a rule from docs of a different ShardingSphere version where the bundled channel set differs.

Related errors

• 1
• Either key generator name or algorithm segment must be provi
• SPI:1
• SPI:1
• Source storage units have different database types

AI-assisted analysis of apache/shardingsphere@e952770a21 (2026-08-14). Data as JSON: /api/errors/ee7d0868dcfc3912. Report an issue: GitHub.