endToEnd feature doesn't produce message to topic at first kminion start

[alarex]

Using kminion v2.2.8 as deployment and deploying it with strimzi kafka 3.7.0 to k8s. When topic management is enabled, kminion creates topic and consumer group, but does't produce anything until restarted, after restart working as expected.
Tested other versions down to 2.2.3, lower versions create topic and write init messages, after that same behavior, no messages until deployment restart.

Tried to create topic manually, if topic management enabled, kminion stiil make attempt to create topic, and then nothing until restart.

When i disable topic management and create topic manually, kminion work as expected at first run

Logs from first deploy when topic management is enabled:

{"level":"info","ts":"2024-07-02T09:53:27.112Z","logger":"main","msg":"successfully connected to kafka cluster"}
{"level":"debug","ts":"2024-07-02T09:53:27.113Z","logger":"main.e2e","msg":"validating end-to-end topic..."}
{"level":"debug","ts":"2024-07-02T09:53:27.113Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"2","bytes_written":47,"write_wait":0.000017434,"time_to_write":0.000046283,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.115Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"2","bytes_read":486,"read_wait":0.000052535,"time_to_read":0.002121352,"err":null}
{"level":"info","ts":"2024-07-02T09:53:27.115Z","logger":"main.e2e","msg":"e2e topic does not exist, creating it...","topic_name":"kminion-end-to-end","partitions_per_broker":1,"replication_factor":1,"broker_count":3,"total_partitions":3}
{"level":"debug","ts":"2024-07-02T09:53:27.115Z","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-0.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"0"}
{"level":"debug","ts":"2024-07-02T09:53:27.123Z","logger":"main.e2e_hooks","msg":"kafka connection succeeded","host":"kf-kminion-test-staging-kf-kminion-test-staging-broker-0.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc","broker_id":0,"dial_duration_ms":8}
{"level":"debug","ts":"2024-07-02T09:53:27.123Z","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-0.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"0"}
{"level":"debug","ts":"2024-07-02T09:53:27.123Z","logger":"main.kgo_client","msg":"issuing api versions request","broker":"0","version":3}
{"level":"debug","ts":"2024-07-02T09:53:27.123Z","logger":"main.kgo_client","msg":"wrote ApiVersions v3","broker":"0","bytes_written":35,"write_wait":0.000006043,"time_to_write":0.000142217,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.129Z","logger":"main.kgo_client","msg":"read ApiVersions v3","broker":"0","bytes_read":475,"read_wait":0.000044843,"time_to_read":0.005659602,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.129Z","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-0.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"0"}
{"level":"debug","ts":"2024-07-02T09:53:27.129Z","logger":"main.kgo_client","msg":"wrote CreateTopics v6","broker":"0","bytes_written":147,"write_wait":0.014105967,"time_to_write":0.000040174,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.240Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"0","bytes_written":47,"write_wait":0.000034667,"time_to_write":0.000157164,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.241Z","logger":"main.kgo_client","msg":"read CreateTopics v6","broker":"0","bytes_read":1108,"read_wait":0.000031119,"time_to_read":0.111869374,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.241Z","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"1"}
{"level":"debug","ts":"2024-07-02T09:53:27.245Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"0","bytes_read":486,"read_wait":0.00015292,"time_to_read":0.005544891,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.246Z","logger":"main.kgo_client","msg":"immediate metadata update had inner errors, re-updating","errors":"UNKNOWN_TOPIC_OR_PARTITION{__consumer_offsets}","update_after":0.25}
{"level":"debug","ts":"2024-07-02T09:53:27.253Z","logger":"main.e2e_hooks","msg":"kafka connection succeeded","host":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc","broker_id":1,"dial_duration_ms":11}
{"level":"debug","ts":"2024-07-02T09:53:27.253Z","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"1"}
{"level":"debug","ts":"2024-07-02T09:53:27.253Z","logger":"main.kgo_client","msg":"issuing api versions request","broker":"1","version":3}
{"level":"debug","ts":"2024-07-02T09:53:27.253Z","logger":"main.kgo_client","msg":"wrote ApiVersions v3","broker":"1","bytes_written":35,"write_wait":0.000017597,"time_to_write":0.000056001,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.261Z","logger":"main.kgo_client","msg":"read ApiVersions v3","broker":"1","bytes_read":475,"read_wait":0.000051631,"time_to_read":0.008207835,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.261Z","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"1"}
{"level":"debug","ts":"2024-07-02T09:53:27.262Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"1","bytes_written":47,"write_wait":0.020249493,"time_to_write":0.000154321,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.271Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"1","bytes_read":486,"read_wait":0.000101679,"time_to_read":0.009705721,"err":null}
{"level":"info","ts":"2024-07-02T09:53:27.271Z","logger":"main.e2e","msg":"initializing consumer and waiting until it has received the first record batch"}
{"level":"info","ts":"2024-07-02T09:53:27.272Z","logger":"main.e2e","msg":"starting to consume end-to-end topic","topic_name":"kminion-end-to-end","group_id":"kminion-end-to-end-fa4b069a-3d88-4872-802a-c11b21cd1197"}
{"level":"debug","ts":"2024-07-02T09:53:27.329Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"2","bytes_written":47,"write_wait":0.000083413,"time_to_write":0.000303764,"err":null}
{"level":"debug","ts":"2024-07-02T09:53:27.330Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"2","bytes_read":486,"read_wait":0.000248975,"time_to_read":0.000940289,"err":null}

Logs after kminion deployment restart:

{"level":"debug","ts":"2024-07-02T10:05:49.669Z","logger":"main.kafka_service","msg":"successfully connected to kafka cluster","advertised_broker_count":3,"topic_count":5,"controller_id":2,"kafka_version":"at least v3.6"}
{"level":"info","ts":"2024-07-02T10:05:49.669Z","logger":"main","msg":"successfully connected to kafka cluster"}
{"level":"debug","ts":"2024-07-02T10:05:49.669Z","logger":"main.e2e","msg":"validating end-to-end topic..."}
{"level":"debug","ts":"2024-07-02T10:05:49.669Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"2","bytes_written":47,"write_wait":0.000007317,"time_to_write":0.000100092,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.670Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"2","bytes_read":564,"read_wait":0.000058359,"time_to_read":0.000606991,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.670Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"2","bytes_written":47,"write_wait":0.000007399,"time_to_write":0.000037833,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.671Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"2","bytes_read":564,"read_wait":0.000038534,"time_to_read":0.000503031,"err":null}
{"level":"info","ts":"2024-07-02T10:05:49.671Z","logger":"main.e2e","msg":"initializing consumer and waiting until it has received the first record batch"}
{"level":"info","ts":"2024-07-02T10:05:49.671Z","logger":"main.kgo_client","msg":"immediate metadata update triggered","why":"forced load because we are producing to a topic for the first time"}
{"level":"info","ts":"2024-07-02T10:05:49.671Z","logger":"main.e2e","msg":"starting to consume end-to-end topic","topic_name":"kminion-end-to-end","group_id":"kminion-end-to-end-f4edccc9-32fa-4499-b5f4-84fea9e3a68a"}
{"level":"info","ts":"2024-07-02T10:05:49.671Z","logger":"main.kgo_client","msg":"producing to a new topic for the first time, fetching metadata to learn its partitions","topic":"kminion-end-to-end"}
{"level":"debug","ts":"2024-07-02T10:05:49.674Z","logger":"main.e2e_hooks","msg":"kafka connection succeeded","host":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc","broker_id":1,"dial_duration_ms":8}
{"level":"debug","ts":"2024-07-02T10:05:49.674Z","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"1"}
{"level":"debug","ts":"2024-07-02T10:05:49.674Z","logger":"main.kgo_client","msg":"issuing api versions request","broker":"1","version":3}
{"level":"debug","ts":"2024-07-02T10:05:49.675Z","logger":"main.kgo_client","msg":"wrote ApiVersions v3","broker":"1","bytes_written":35,"write_wait":0.000014843,"time_to_write":0.000051426,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.682Z","logger":"main.kgo_client","msg":"read ApiVersions v3","broker":"1","bytes_read":475,"read_wait":0.000047556,"time_to_read":0.007340048,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.682Z","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"1"}
{"level":"debug","ts":"2024-07-02T10:05:49.682Z","logger":"main.kgo_client","msg":"wrote FindCoordinator v3","broker":"1","bytes_written":80,"write_wait":0.016164308,"time_to_write":0.000102471,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.682Z","logger":"main.kgo_client","msg":"wrote Metadata v9","broker":"1","bytes_written":47,"write_wait":0.005584565,"time_to_write":0.000073832,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"read FindCoordinator v3","broker":"1","bytes_read":143,"read_wait":0.000137425,"time_to_read":0.007741936,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-1.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"1"}
{"level":"debug","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"read Metadata v9","broker":"1","bytes_read":564,"read_wait":0.007836293,"time_to_read":0.000221633,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"metadata refresh has identical topic partition data","topic":"kminion-end-to-end","partition":0,"leader":1,"leader_epoch":0}
{"level":"debug","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"metadata refresh has identical topic partition data","topic":"kminion-end-to-end","partition":1,"leader":2,"leader_epoch":0}
{"level":"debug","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"metadata refresh has identical topic partition data","topic":"kminion-end-to-end","partition":2,"leader":0,"leader_epoch":0}
{"level":"info","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"done waiting for metadata for new topic","topic":"kminion-end-to-end"}
{"level":"info","ts":"2024-07-02T10:05:49.690Z","logger":"main.kgo_client","msg":"initializing producer id"}
{"level":"debug","ts":"2024-07-02T10:05:49.691Z","logger":"main.kgo_client","msg":"wrote InitProducerID v4","broker":"2","bytes_written":38,"write_wait":0.000011606,"time_to_write":0.000115139,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.691Z","logger":"main.kgo_client","msg":"read InitProducerID v4","broker":"2","bytes_read":26,"read_wait":0.000048221,"time_to_read":0.000401465,"err":null}
{"level":"info","ts":"2024-07-02T10:05:49.691Z","logger":"main.kgo_client","msg":"producer id initialization success","id":2,"epoch":0}
{"level":"debug","ts":"2024-07-02T10:05:49.691Z","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-2.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"2"}
{"level":"debug","ts":"2024-07-02T10:05:49.691Z","logger":"main.kgo_client","msg":"opening connection to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-0.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"0"}
{"level":"debug","ts":"2024-07-02T10:05:49.693Z","logger":"main.e2e_hooks","msg":"kafka connection succeeded","host":"kf-kminion-test-staging-kf-kminion-test-staging-broker-2.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc","broker_id":2,"dial_duration_ms":1}
{"level":"debug","ts":"2024-07-02T10:05:49.693Z","logger":"main.kgo_client","msg":"connection opened to broker","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-2.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"2"}
{"level":"debug","ts":"2024-07-02T10:05:49.693Z","logger":"main.kgo_client","msg":"connection initialized successfully","addr":"kf-kminion-test-staging-kf-kminion-test-staging-broker-2.kf-kminion-test-staging-kafka-brokers.kafka-mks-test.svc:9092","broker":"2"}
{"level":"debug","ts":"2024-07-02T10:05:49.693Z","logger":"main.kgo_client","msg":"wrote Produce v8","broker":"2","bytes_written":145,"write_wait":0.001524208,"time_to_write":0.000130014,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.695Z","logger":"main.kgo_client","msg":"read Produce v8","broker":"2","bytes_read":76,"read_wait":0.000051515,"time_to_read":0.002352019,"err":null}
{"level":"debug","ts":"2024-07-02T10:05:49.695Z","logger":"main.kgo_client","msg":"produced","broker":"2","to":"kminion-end-to-end[1{0=>1}]"}

[bachmanity1]

Hi @alarex, this is probably related to this #268.