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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
