multishard_writer: Abort the queue attached to consumers when producer fails
multishard_writer: Abort the queue attached to consumers when producer fails

We have this in multishard_writer:

future<uint64_t> multishard_writer::operator()() {
    return distribute_mutation_fragments().finally([this] {
        return wait_pending_consumers();
    }).then([this] {
        return _consumed_partitions;
    });
}

The wait_pending_consumers which waits for the consumers to finish is
called even when distribute_mutation_fragments fails.

When distribute_mutation_fragments fails and the failure is due to the
producer fails, consumers can wait for data which will never come because
the producer has failed already. This can cause a deadlock.

To fix, when distribute_mutation_fragments fails, we should abort the
queues that are attached to the readers used by the consumers.

Fixes #6241

scylladb/scylla,scylladb/scylla,avikivity/scylla,avikivity/scylla,avikivity/scylla,scylladb/scylla,scylladb/scylla