apache / apache/iggy

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

Open
#4,097 0 comments 0 reactions 1 assignee Claimed by @mlevkov View on GitHub
bug connectors rust
Dominant language
Rust
Stars
4.9k
Forks
432
Avg merge
2d 10h
Merged PRs (30d)
173

Description

## 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.

### Contribution

- [x] I'm willing to submit a pull request to fix this bug

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.