Kafkajs: The producer attempted a transactional operation in an invalid state

[eliw00d]

So, using the example given here, I tried to set up a transactional producer.

client:

const client = new Kafka({
  clientId: 'my-client',
  brokers: ['my-broker'],
})
const producer = client.producer({ 
  maxInFlightRequests: 1, 
  idempotent: true,
  transactionalId: 'my-transaction'
})

transaction:

const  transaction = await producer.transaction()

try {
  await transaction.send({ topic, messages })

  await transaction.commit()
} catch (e) {
  await transaction.abort()
}

As long as I send and commit without issue, it works. However, if any error is thrown, and abort is called, I get the following error:

[Connection] Response EndTxn(key: 26, version: 0)
    broker: "my-broker"
    clientId: "my-client"
    error: "The producer attempted a transactional operation in an invalid state"
    correlationId: 1
    size: 10

```
{
"type": "KafkaJSProtocolError",
"message": "The producer attempted a transactional operation in an invalid state",
"stack":
KafkaJSProtocolError: The producer attempted a transactional operation in an invalid state
at createErrorFromCode (/Users/eliw00d/Documents/GitHub/my-project/node_modules/kafkajs/src/protocol/error.js:536:10)
at Object.parse (/Users/eliw00d/Documents/GitHub/my-project/node_modules/kafkajs/src/protocol/requests/endTxn/v0/response.js:24:11)
at Connection.send (/Users/eliw00d/Documents/GitHub/my-project/node_modules/kafkajs/src/network/connection.js:306:35)
at process._tickCallback (internal/process/next_tick.js:68:7)
"name": "KafkaJSProtocolError",
"retriable": false,
"code": 48
}

To better illustrate the issue, I even tried:
```javascript
const transaction = await producer.transaction()
await transaction.abort()

with the same result.

Is this a bug or am I missing something?

[eliw00d]

Actually, I realized that this is probably intentional since there would be nothing to abort if send was never called. However, in our case, the try/catch wraps a database transaction and we want to ensure that if the database transaction fails we abort or cancel the producer transaction as well.

Maybe this could be a feature request for something like transaction.isAbortable that we can call before transaction.abort to avoid this error? Like so:

const  transaction = await producer.transaction()
try {
  // insert code that will create the messages here
  await transaction.send({ topic, messages })
  await transaction.commit()
} catch (e) {
  // only abort if we called transaction.send
  if (transaction.isAbortable()) {
    await transaction.abort()
  }
}

That way, if we never call transaction.send, we won't crash and burn.

[eliw00d]

Update

I noticed this in the logs:

[Producer] Transaction state transition UNINITIALIZED --> READY
[Producer] Transaction state transition READY --> TRANSACTING
[Producer] Transaction state transition TRANSACTING --> COMMITTING

Somehow it went from TRANSACTING to COMMITTING but I never called await transaction.commit()...

[eliw00d]

@tulios I mentioned in the PR that it didn't actually fix the issue. Could we reopen this? I still get the same error when aborting before sending. The PR only fixed the transitionTo (which does say TRANSACTING --> ABORTING now).

[tulios]

Yeah, sorry. Github closed it after the merge.

[jasonr44]

Hey all,

Anyone actively working on this? Trying to look into it a little on my end as it has become an issue in a current project.

[konapun]

(Disclaimer: I'm not a kafkajs maintainer) I've looked into this a bit and it looks like the error is caused by the kafka backend not transitioning to a transacting state until producer.send is called so if the kafkajs client attempts to abort a transaction before the send is attempted you'll receive that error.

To get around this I'm wrapping the abort in a try/catch and swallowing the error if its code is 48:

try {
    transaction.abort()
} catch (err) {
    if (err.code !== 48) {
        throw err
    }
}

This works for now but I would like to see the actual bug get fixed in kafkajs.

[tulios]

Hey @ankon thanks very much for the workaround. @jasonr44 we will work on this in October, right now the whole team is off on vacation.

[tulios]

@eliw00d I like your proposal of a transaction.isAbortable(), I will check how the Java client work with this use-case and I can start a PR.

[plameniv]

Has there been any movement on the transaction.isAbortable() idea?

[Nevon]

I don't think anyone is actively working on implementing isAbortable. Personally I don't really like the idea of isAbortable, and would rather just make abort not make the abort request if no sends have been made, or handle the error internally. I don't think userland code should have to worry about that.

I tried looking into what the Java client does, but I haven't been able to find any special handling of abortTransaction in there, nor can I find any mention of when it is valid to send an EndTxn request. Looking at example Java code, it looks like it should be fine to call abortTransaction at any time.

What we call abort is an EndTxn request with a transaction_result of 0 (meaning abort), and at least in my mind it should be valid to call EndTxn as soon as you have created a transaction (i.e. made a initProducerId request). I suspect that this behavior is actually a bug of some kind, but I have only done a cursory investigation so far.