Add timestamp field to the kafka and kafka_franz outputs.
#2665
Conversation
| Tag: "latest", | ||
| Hostname: "redpanda", | ||
| ExposedPorts: []string{"9092"}, | ||
| PortBindings: map[docker.Port][]docker.PortBinding{ | ||
| "9092/tcp": {{HostIP: "", HostPort: kafkaPortStr}}, | ||
| }, | ||
| Cmd: []string{ | ||
| "redpanda", "start", "--smp 1", "--overprovisioned", "", |
There was a problem hiding this comment.
This doesn't seem to work anymore, at least not via Colima on an M3. I also ran all the Kafka integration tests after these changes on amd64 with Linux and standard Docker and they passed successfully.
| "start", | ||
| "--node-id 0", | ||
| "--mode dev-container", | ||
| "--set rpk.additional_start_flags=[--reactor-backend=epoll]", |
There was a problem hiding this comment.
This was super-annoying, since it sometimes starts successfully if I don't set this option, but not always. Details here: redpanda-data/redpanda#17104
| "redpanda", | ||
| "start", | ||
| "--node-id 0", | ||
| "--mode dev-container", |
There was a problem hiding this comment.
Docs here: https://docs.redpanda.com/current/reference/rpk/rpk-redpanda/rpk-redpanda-mode/#development-mode Among other things, it sets --overprovisioned to true.
> redpanda start --mode help
...
Mode uses well-known configuration properties for development or tests
environments:
--mode dev-container
Bundled flags:
* --overprovisioned
* --reserve-memory 0M
* --check=false
* --unsafe-bypass-fsync
Bundled cluster properties:
* auto_create_topics_enabled: true
* group_topic_partitions: 3
* storage_min_free_bytes: 10485760 (10MiB)
* topic_partitions_per_shard: 1000
* fetch_reads_debounce_timeout: 10
* group_initial_rebalance_delay: 0
* log_segment_size_min: 1
* write_caching_default: true
Fixes #398. Note: In #398, the broker `CreateTime` mode is mentioned and I believe this is the mode that's being used by both the `kafka` and `kafka_franz` output. See the comment from Sarama here: https://github.com/IBM/sarama/blob/d2246cc9b9df113f3ffed7c442fd81e8b9cf736e/async_producer.go#L181-L192. Also, see this issue in franz-go: twmb/franz-go#166. Signed-off-by: Mihai Todor <todormihai@gmail.com>
mihaitodor
force-pushed
the
mihaitodor-add-timestamp-field-kafka
branch
from
ee59587
to
637f64c
Compare
| @@ -356,3 +364,76 @@ input: | |||
| integration.StreamTestOptVarSet("VAR1", ""), | |||
| ) | |||
| } | |||
|
|
|||
| func TestIntegrationKafkaOutputFixedTimestamp(t *testing.T) { | |||
There was a problem hiding this comment.
It would be nice to maybe avoid adding extra tests. I was thinking to introduce a new StreamTestReadsMetadata() which writes something to the output and then launches the input similar to StreamTestOpenCloseIsolated() and then checks if the metadata of the message read from the input contains the metadata keys and values which were passed via a (variadic) parameter.
| consumer_group: "blobfish" | ||
| processors: | ||
| - mapping: | | ||
| root = if metadata("kafka_timestamp_unix") != 666 { "error: invalid timestamp" } |
There was a problem hiding this comment.
This is a bit messy... Would you be open to updating messageMatch() here to also check that for the received message.Part ErrorGet() returns a nil error? I'd like to use throw("invalid timestamp") instead.
There was a problem hiding this comment.
Yeah I think that'd be fine, it's a niche that any existing tests would break as a result and it's not a big deal to have minor behavioural changes in the testing suite.
There was a problem hiding this comment.
Looks great, thanks @mihaitodor, and extra ❤️ for fixing up the integration tests.
|
Thanks for this <3 For redis streams > kafka |
Fixes #398.
Note: In #398, the broker
CreateTimemode is mentioned and I believe this is the mode that's being used by both thekafkaandkafka_franzoutput. See the comment from Sarama here: https://github.com/IBM/sarama/blob/d2246cc9b9df113f3ffed7c442fd81e8b9cf736e/async_producer.go#L181-L192. Also, see this issue in franz-go: twmb/franz-go#166.