Runtime returning without completing all tasks despite using block_on in Tokio

I am writing a multi-threaded concurrent Kafka producer using Rust and Tokio. The project has 2 modes, an interactive mode that runs in an infinite loop and a file mode which takes a file as an argument and then reads the file and sends these messages to Kafka via multiple threads. Interactive mode works fine! but file mode has issues.

To achieve this, I had initially started with Rayon, but then switched to a more flexible runtime; tokio. Now, I am able to parallelize the task of sending data over a specified number of threads within tokio, however, it seems that runtime is getting dropped before all messages are produced. Here is my code:

pub fn worker(brokers: String, f: File, t: usize, topic: Arc<String>) {
    let reader = BufReader::new(f);
    let mut rt = runtime::Builder::new()
        .threaded_scheduler()
        .core_threads(t)
        .build()
        .unwrap();

    let producers: Arc<Vec<Mutex<BaseProducer>>> = Arc::new(
        (0..t)
            .map(|_| get_producer(&brokers))
            .collect::<Vec<Mutex<BaseProducer>>>(),
    );

    let acounter = atomic::AtomicUsize::new(0);

    let _results: Vec<_> = reader
        .lines()
        .map(|line| line.unwrap())
        .map(move |line| {
            let prods = producers.clone();
            let tp = topic.clone();
            let cnt = acounter.swap(
                (acounter.load(atomic::Ordering::SeqCst) + 1) % t,
                atomic::Ordering::SeqCst,
            );
            rt.block_on(async move {
                match prods[cnt]
                    .lock()
                    .unwrap()
                    .send(BaseRecord::to(&(*tp)).payload(&line).key(""))
                {
                    Ok(_) => (),
                    Err(e) => eprintln!("{:?}", e),
                };
            })
        })
        .collect();
}

fn get_producer(brokers: &String) -> Mutex<BaseProducer> {
    Mutex::new(
        BaseProducer::from_config(
            ClientConfig::new()
                .set("bootstrap.servers", &brokers)
                .set("message.timeout.ms", "5000"),
        )
        .expect("Producer creation error"),
    )
}

As a high-level walkthrough: I create mutable producers equal to the number of threads specified and every task within this thread will use one of these producers. The file is read line by line sequentially and every line is moved into the closure that produces it as a message to Kafka.

The code works fine, for the most part, but there are issues related to the runtime exiting without completing all tasks, even when I am using the block_on function in the runtime. Which is supposed to block until the future is complete (Async block in my case here).

I believe the issue is that the issue is with runtime getting dropped without all the threading within Tokio exiting successfully.

I tried reading a file with this approach habing 100,000 records, on a single thread, I was able to produce 28,000 records. On 2 threads, close to 46,000 records. And while utilising all 8 cores of my CPU, I was getting 99,000-100,000 messages indeterministically.

I have checked several answers on SO, but none help in my case. I also read through the documentation of tokio::runtime::Runtime here and tried to use spawn and then use futures::future::join, but that didn't work either.

Any help is appreciated!