feat: add allow_create_stream to avoid create nats stream by mistake

[tabVersion]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

resolve #21797

per request by poc user

What's changed and what's your intention?

Problem

The NATS connector currently creates streams automatically when they don't exist, which can lead to unintended stream creation in production environments. This poses security and operational risks:
Streams may be created with default configurations that don't match production requirements
Typos in stream names could result in unwanted streams being created
No explicit control over when RisingWave should have stream creation permissions

Solution

This PR introduces a new boolean configuration parameter allow_create_stream that provides explicit control over stream creation behavior:
Default: allow_create_stream = false - RisingWave will NOT create streams automatically
Explicit permission: Users must set allow_create_stream = true to enable stream creation
Clear error messaging: When a stream doesn't exist and creation is disabled, users get a helpful error message

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

[Copilot]

Pull Request Overview

This PR adds an allow_create_stream flag to the NATS connector options to prevent accidental creation of JetStream streams by default.

• Introduces allow_create_stream in source and sink YAML definitions with a default of false.
• Extends NatsCommon to parse the new flag and defaults it to false.
• Updates build_or_get_stream to error when a stream is missing and allow_create_stream is not set, and refreshes integration tests and docs accordingly.

Reviewed Changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated no comments.

File	Description
src/connector/with_options_source.yaml	Add allow_create_stream property to source connector YAML
src/connector/with_options_sink.yaml	Add allow_create_stream property to sink connector YAML
src/connector/src/connector_common/common.rs	Add allow_create_stream field and enforce guard in stream builder
integration_tests/nats/*	Update SQL tests and README to include allow_create_stream

Comments suppressed due to low confidence (5)

src/connector/src/connector_common/common.rs:956

• [nitpick] Consider renaming stream_str to stream_name for clarity and consistency with other parts of the codebase.

        stream_str: String,

src/connector/src/connector_common/common.rs:838

• Add a rustdoc comment explaining the purpose of allow_create_stream and its default behavior (false) so users understand its effect without reading the implementation.

    pub allow_create_stream: bool,

src/connector/src/connector_common/common.rs:968

• Consider adding a unit test for build_or_get_stream that verifies it errors when allow_create_stream is false and the stream does not exist.

        if !self.allow_create_stream {

src/connector/with_options_source.yaml:695

• [nitpick] Using Default::default in YAML may be ambiguous to users of generated docs; consider specifying default: false explicitly for clarity.

    default: Default::default

src/connector/src/connector_common/common.rs:969

• The error message includes the stream name and backticks around the config field; ensure this matches any consumers or tests that expect a specific substring.

            return Err(anyhow!(

[xiangjinwu]

Will existing sources created before this PR also default to false, leading to them unable to create the stream automatically?

Similar issue: #22206

[tabVersion]

The issue happens with little possibility.

For most existing streaming job, once they start the source, there must be a stream running. If the cluster experience a recovery, it will not hit the check here, it gets the stream directly without creating one.

One exception is, drop the stream and rely on RisingWave to recreate it. I think it is a wrong usage and we can ignore the case. Because creating a stream materializes some data on Nats server, this should be handled with caution.

[github-actions]

✅ Cherry-pick PRs (or issues if encountered conflicts) have been created successfully to all target branches.