[I] connectors: a source with two [[streams]] entries produces to only one [iggy]

["mlevkov (via GitHub)" <gi...@apache.org>]

mlevkov opened a new issue, #4097:
URL: https://github.com/apache/iggy/issues/4097

   ## What happens
   
   `setup_source_producer` in `core/connectors/runtime/src/source.rs` loops over every
   `[[streams]]` entry, builds a producer and an encoder for each one, calls `init()` on each,
   and then keeps only the last pair:
   
   ```rust
   let mut last_producer = None;
   let mut last_encoder = None;
   for stream in config.streams.iter() {
       // ...
       let producer = iggy_client.producer(&stream.stream, &stream.topic)?/* ... */.build();
       producer.init().await?;
       // ...
       last_encoder = Some(encoder);
       last_producer = Some(producer);
   }
   ```
   
   `source_forwarding_loop` takes one `producer` and one `encoder`, so a source configured with
   two streams sends every message to the second one. There is no error and no warning.
   
   `init()` creates a missing stream and topic by default: `create_stream_if_not_exists` and
   `create_topic_if_not_exists` are both `true` in `IggyProducerBuilder::new`, and the source path
   does not turn them off. So every configured stream and topic really is created. They just stay
   empty except the last one. That is what makes this hard to spot. The config looks applied.
   
   ## Why it matters
   
   The sink side does not have this problem. `setup_sink_consumers` collects a consumer for every
   stream and topic into a `Vec` and spawns a task per entry. So `[[streams]]` is a list that works
   on one side and silently drops all but one element on the other.
   
   Messages an operator meant for two topics land on one. Nothing counts it, because as far as the
   runtime is concerned every message was sent successfully.
   
   Every example config in the tree has exactly one `[[streams]]` entry, which is probably why this
   has not been hit.
   
   ## Fix
   
   Two options, and the choice is a real one:
   
   1. Carry all producers and encoders through to `source_forwarding_loop` and fan each batch out
      to all of them. This is what the field name promises and it matches the sink. It also raises
      questions this issue does not settle: whether a partial send failure should NACK the batch,
      and whether state should be saved when only some destinations accepted it.
   2. Refuse a source config with more than one `[[streams]]` entry until fan-out exists. Smaller,
      and it turns silent misrouting into a startup error.
   
   Option 2 is the small honest change. Option 1 is the feature.
   
   ## Provenance
   
   Raised by @hubcio in review on #4064, listed as outside that PR's diff and explicitly
   non-blocking for it. Filed separately so it does not go away when #4064 merges;
   `CONTRIBUTING.md` "Single Purpose" keeps it out of that PR.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org