bug(source): certain properties/options missing after upgrading from v1.5.0 to v1.6.0

[xiangjinwu]

Describe the bug

In #13762, we intended to fix a problematic design where WITH (k0 = v0) FORMAT ... ENCODE ... (k1 = v1) shares a single KV map. The PR tries to maintain backward compatible by detecting legacy sources and copying necessary KV pairs from one position to another (see src/meta/src/manager/catalog/mod.rs). However, it seems not working as intended, and certain KV pairs are lost, causing a source to malfunction after upgrade.

Notably, this include schema registry username and password if they were given after WITH before the upgrade.

Error message/log

No response

To Reproduce

Start v1.5.0 server:

git checkout v1.5.0
./risedev configure # enable minio and etcd
vim risedev.yml # uncomment minio, etcd and compactor
./risedev d

Create a table with source (kafka broker at 51349 and http server at 8000):

create table t with (connector = 'kafka', topic = 'test-topic', properties.bootstrap.server = '127.0.0.1:51349', schema.registry.username = 'foo') format plain encode avro (schema.registry = 'http://127.0.0.1:8000');

Add logging. Restart v1.5.0 to see username is retained:

vim src/connector/src/schema/schema_registry/client.rs +83 # output client_config.username in Client::new
./risedev k
./risedev d

Upgrade to v1.6.0 and see username is lost:

./risedev k
git stash save 'inspect schema registry username'
git checkout v1.6.0
git stash apply
./risedev d

Expected behavior

Sources created in v1.5.0, even with schema.registry.username specified after WITH, shall still send it when calling schema registry API upgrading to v1.6.x.

How did you deploy RisingWave?

RiseDev

The version of RisingWave

from the v1.5.0 tag to v1.6.0 tag

Additional context

Dummy schema served by http server on 8000: subjects/test-topic-value/versions/latest

{
  "id": 71,
  "subject": "test-topic-value",
  "version": 2,
  "schema": "{\"type\":\"record\",\"name\":\"TestVal\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}"
}

cc @Rossil2012 @tabVersion

[xiangjinwu]

Impact

Infinite retry loop with the following error:
https://github.com/risingwavelabs/risingwave/blob/v1.6.0/src/stream/src/executor/source/source_executor.rs#L237

Root cause

The compatibility logic in CatalogManager (src/meta/src/manager/catalog/mod.rs) is not enough. There is another copy of PbStreamSourceInfo stored in FragmentManager which is actually used by SourceExecutorBuilder.

Potential fix

(unfortunately we already have 3 different flavors in use):

• Similar to sink format encode compatibility logic, avoid mutating stored protobuf but handle it during PbStreamSourceInfo decoding.
• Similar to time zone expr context compatibility logic, avoid mutating stored protobuf but handle it during FragmentManager::new loading.
• Similar to source rate limit or current CatalogManager::init compatibility logic, also mutate stored protobuf of FragmentManager and commit it to meta.