feat(meta): enhance inject source offsets with better validation and feedback

- Validate split offsets against runtime actor assignments (SharedActorInfos) instead of enumerator to support adaptive splits correctly.
- Return the list of actually applied split IDs in the RPC response.
- Fail the executor explicitly if offset injection fails (e.g. invalid format) to ensure consistency.
- Update CLI to show applied splits to the user.
- Fix e2e test offset calculation to account for start_offset behavior.

Co-Authored-By: Claude Sonnet 4.5 <[email protected]>

Author: tabVersion

e2e_test/source_inline/kafka/alter/alter_source_properties_safe.slt

@@ -112,7 +112,10 @@ SELECT COUNT(*) FROM mv_test;
 9
 
 # Get Kafka offsets with skip_last=4 to go back 4 messages
-# This will inject offset = high_watermark - 4, causing re-consumption of last 4 messages
+# The script calculates offset = high_watermark - 4 - 1.
+# Since inject-source-offsets treats the value as "last consumed offset",
+# consumption will resume from (high_watermark - 4 - 1) + 1 = high_watermark - 4.
+# This causes re-consumption of the last 4 messages.
 system ok
 python3 e2e_test/source_inline/kafka/alter/get_kafka_offsets.py \
     --topic test_safe_update \

e2e_test/source_inline/kafka/alter/get_kafka_offsets.py

@@ -31,8 +31,9 @@ def get_kafka_offsets(broker, topic, skip_last_n=2):
             tp = TopicPartition(topic, partition_id)
             low, high = consumer.get_watermark_offsets(tp)
             # Set offset to skip last N messages
-            # This simulates moving forward in the stream
-            new_offset = max(0, high - skip_last_n)
+            # inject_source_offsets expects "last_seen_offset", so start_offset = offset + 1.
+            # We subtract 1 to ensure we start reading from (high - skip_last_n).
+            new_offset = high - skip_last_n - 1
             offsets[str(partition_id)] = str(new_offset)
 
         return offsets