What causes "{license Feature::RedisSinkStream unavailable error}" in risingwavelabs/risingwave?

Install or renew a RisingWave enterprise license that includes the Redis stream sink feature.

risingwavelabs/risingwave · error · SinkError

{license Feature::RedisSinkStream unavailable error}

Error message

{license Feature::RedisSinkStream unavailable error}

What it means

The Redis stream value type (`value type = 'stream'`) is gated behind a licensed feature, `Feature::RedisSinkStream`. During `validate`, the license check is run via `check_available()`; when the deployment lacks a valid license covering this feature, validation fails with the license feature-unavailable error message.

Solutions

Install or renew a RisingWave enterprise license that includes the Redis stream sink feature.
Switch `value type` to `string`, `geo`, or `pubsub`, which are not license gated.
Contact RisingWave sales/support if the license should include this feature but the check still fails.

Example fix

// before
WITH (connector='redis', value type='stream', stream='alerts')  // unlicensed
// after
WITH (connector='redis', value type='pubsub', channel='alerts')

Defensive patterns

Strategy: fallback

Validate before calling

// Query license status before creating a stream-type redis sink
const licensed = await checkFeature('redis_sink_stream');
if (!licensed) useAlternativeValueType('pubsub');

Try / catch

try { await createSink(cfg); } catch (e) { if (/unavailable|license/i.test(String(e))) { cfg['value type'] = 'string'; await createSink(cfg); } else throw e; }

Prevention

Verify the enterprise license covers Redis stream sink before designing around it
Monitor license expiry in deployment automation
Prefer non-gated value types in dev/test environments

When it happens

Trigger: Create a sink with `value type = 'stream'` on an unlicensed, expired-license, or free-tier RisingWave deployment; the check is also hit in validation even before any data flows.

Common situations: Developers testing stream sinks locally without a license key; enterprises whose license expired or excludes the Redis stream sink feature; the preceding `tracing::error!("test:for bug")` indicates this branch was also under debugging.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/51de6a6e30629947. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/redis.rs:451

                        && let Some(channel_column_type) = all_map.get(channel_column)
                        && (channel_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{CHANNEL_COLUMN}` must be set to `varchar`"
                        )));
                    }

                    let value_format =
                        self.format_desc.options.get(VALUE_FORMAT).ok_or_else(|| {
                            SinkError::Config(anyhow!("Cannot find `{VALUE_FORMAT}`"))
                        })?;
                    TemplateStringEncoder::check_string_format(value_format, &all_map)?;
                }
                Some(REDIS_VALUE_TYPE_STREAM) => {
                    tracing::error!("test:for bug");
                    risingwave_common::license::Feature::RedisSinkStream
                        .check_available()
                        .map_err(|e| anyhow::anyhow!(e))?;
                    let stream = self.format_desc.options.get(STREAM);
                    let stream_column = self.format_desc.options.get(STREAM_COLUMN);
                    if (stream.is_none() && stream_column.is_none())
                        || (stream.is_some() && stream_column.is_some())
                    {
                        return Err(SinkError::Config(anyhow!(
                            "Please specific either `{STREAM}` or `{STREAM_COLUMN}`. They are mutually exclusive options."
                        )));
                    }

                    if let Some(stream_column) = stream_column
                        && let Some(stream_column_type) = all_map.get(stream_column)
                        && (stream_column_type != &DataType::Varchar)
                    {
                        return Err(SinkError::Config(anyhow!(
                            "`{STREAM_COLUMN}` must be set to `varchar`"
                        )));
                    }

View on GitHub (pinned to 6469eb736d)