Transactions: ReadyNotAcked error

I'm trying to work with transactions so that I can consume a message and publish new messages in response, all in a single tx. I get an error when I try to init the transaction, and an error when I try to start the tx. Unfortunately, there isn't much in the way of useful info in the error messages, other than:

```json
{
"message":"Operation not valid in state ReadyNotAcked",
"code":-172,
"errno":-172,
"origin":"kafka"
}
```

**Environment Information**
- Ubuntu: 22+
- Node: 19.4.0
- node-rdkafka: 2.14.5

**code**
Sorry, this is quite long:

```javascript
import { v4 as uuid } from "uuid";
import Kafka from "node-rdkafka";
import { log } from "@bryandollery/simple-logging";

const BusUrl = "localhost:9092";
const Group = `tenants.router.1`;

log.setCtx('test');

const main = async _ => {
try {
const producer = await producerConnect();
const consumer = await consumerConnect();
await consume(consumer, producer);
} catch (e) {
log.error('Event adaptor: main', { cause: e });
}
};

const handle = async (message, consumer, producer) => handleEvent(JSON.parse(message.value.toString()), consumer, producer);

const handleEvent = async (event, consumer, producer) => {
const cloudEventData = event.data;

try {
producer.beginTransaction(err => log.debug("Started transaction", err));

publishResults([], [], cloudEventData);
producer.sendOffsetsToTransaction(event.offset, consumer, 500, err => log.debug("Sent offsets to transaction error", err));
producer.commitTransaction(500, err => log.debug("Committed transaction", err));
} catch (e) {
log.error("Error handling event", event, { ...e });
producer.abortTransaction(500, err => log.debug("Aborted transaction error", err));
}
};

const publishResults = async (event, commands, cloudEvent) => {
for (const thing of [...event, ...commands]) {
const newEvent = {
specversion: "1.0",
datacontenttype: "application/json",
tenantId: cloudEvent.tenantId,
type: thing.type,
subtype: thing.subtype,
source: `tenants`,
subject: thing.id,
time: new Date().toISOString(),
correlationId: cloudEvent?.correlationId ? `${cloudEvent.correlationId}-${uuid()}` : uuid(),
data: thing
};

await producer.produce(`${`${cloudEvent.tenantId}-tenants`}`, null, Buffer.from(JSON.stringify(newEvent)));
}
};

const getName = tenant => `${tenant}-tenants`;

const consumerConnect = _ => {
const consumer = new Kafka.KafkaConsumer({
'metadata.broker.list': BusUrl,
'group.id': Group,
'enable.auto.commit': false
});

consumer.on('event.log', logEntry => {
log.debug('Consumer: log', logEntry);
});

consumer.connect();

return new Promise((resolve, reject) => {
consumer.on('ready', function (arg) {
log.debug('Consumer: ready', arg);
resolve(consumer);
});

consumer.on('event.error', function (err) {
log.error('Consumer: error', err);
reject(err);
});
});

};

const consume = async (consumer, producer) => {
consumer.subscribe(["tenants-1000"]);
consumer.on("data", async message => await handle(message, consumer, producer));
consumer.on("warning", warning => log.warn("Consumer warning", warning));
consumer.on("disconnected", event => log.info("Consumer disconnected", event));
consumer.on("error", error => log.error("Consumer error", error));

await consumer.consume();
log.debug(`Consuming`, (await consumer.getMetadata()) || "no metadata");
};

const producerConnect = async _ => {
try {
const producer = new Kafka.Producer({
"bootstrap.servers": BusUrl,
"transactional.id": uuid(),
'debug': 'all'
});

producer.connect();

return new Promise((resolve, reject) => {
producer.on("ready", () => {
log.debug(`Producer: connected`);
producer.initTransactions(500, err => log.error("Producer: initTransactions error", err));
resolve(producer);
});

producer.on('event.error', function (err) {
log.error('Producer: Error', err);
reject(err);
});
});
} catch (e) {
const err = { message: `FATAL ERROR: Event adaptor Producer connection failed`, type: "error", subtype: "fatal", source: "producerConnect", cause: e };
log.error(err);
process.kill(process.pid, "SIGINT");
}
};

main();
```
My package.json only has dependencies on `node-rdkafka`, `uuid`, and `"@bryandollery/simple-logging": "^2.1.0"` and, I run it with `yarn test`.

I use the vscode plugin "Tools for Apache Kafka" to test this with the following file:

```json
PRODUCER keyed-message
topic: tenants-1000
key: 100000
{
"specversion": "1.0",
"datacontenttype": "application/json",
"TenantId": "1000",
"type": "event",
"name": "orderPlaced",
"entity": "tenant",
"source": "test",
"subject": 100000,
"time": "2021-03-03T14:00:00.000Z",
"correlationId": 10,
"data": {
"id": 100000,
"name": "Tenant 1000",
"description": "Tenant 1000",
"created": "2021-03-03T14:00:00.000Z",
"updated": "2021-03-03T14:00:00.000Z"
}
}
```

Startup gives me this output:

```json
{"message":"Producer: connected","ctx":"test","level":"DEBUG","date":1674702576574}
{"message":"Consumer: ready","data":[{"name":"rdkafka#consumer-2"}],"ctx":"test","level":"DEBUG","date":1674702576581}
{"message":"Consuming","data":["no metadata"],"ctx":"test","level":"DEBUG","date":1674702576582}
{"message":"Producer: initTransactions error","data":[null],"ctx":"test","level":"ERROR","date":1674702576583}
```

And, sending the message adds this output:

```json
{"message":"Error handling event","data":[{"specversion":"1.0","datacontenttype":"application/json","TenantId":"1000","type":"event","name":"orderPlaced","entity":"tenant","source":"test","subject":100000,"time":"2021-03-03T14:00:00.000Z","correlationId":10,"data":{"id":100000,"name":"Tenant 1000","description":"Tenant 1000","created":"2021-03-03T14:00:00.000Z","updated":"2021-03-03T14:00:00.000Z"}},{}],"ctx":"test","level":"ERROR","date":1674701794946}
{"message":"Started transaction","data":[{"message":"Operation not valid in state ReadyNotAcked","code":-172,"errno":-172,"origin":"kafka"}],"ctx":"test","level":"DEBUG","date":1674701794948}
{"message":"Aborted transaction","data":[{"message":"Operation not valid in state ReadyNotAcked","code":-172,"errno":-172,"origin":"kafka","isFatal":false,"isRetriable":false,"isTxnRequiresAbort":false}],"ctx":"test","level":"DEBUG","date":1674701794949}
```

Start with the producerConnect and handleEvent entry points in the supplied example, focusing on the initTransactions and beginTransaction callbacks and their ordering. Reproduce the ReadyNotAcked error with the provided transaction flow, then document the cause and the expected transaction behavior; no project file or test is named in the issue.

Written by the indexing model from the issue text.