What was the complete path of optimistic locking from receiving a message from Kafka to updating in the database?

Optimistic locking when processing messages from Kafka and updating the database typically looks like this:

1. Receive a message from Kafka containing data for the update.
2. Extract the entity key and the current record's version (or timestamp) from the message.
3. Execute a database query with a condition that the record's version matches the one in the message (i.e., no one has modified the record since the message was received).
4. If the condition is met, update the record, incrementing the version.
5. If the condition is not met (version conflict), rollback the operation or retry.

Example in Go using SQL:

// msg contains ID and version of the record
func processMessage(msg Message) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()

    res, err := tx.Exec(`UPDATE items SET data = ?, version = version + 1 WHERE id = ? AND version = ?`, msg.Data, msg.ID, msg.Version)
    if err != nil {
        return err
    }

    rowsAffected, err := res.RowsAffected()
    if err != nil {
        return err
    }

    if rowsAffected == 0 {
        // Version conflict — someone else updated the record earlier
        return fmt.Errorf("optimistic lock failed")
    }

    return tx.Commit()
}

Thus, we ensure that the update occurs only if the data has not changed since it was read, preventing data loss during concurrent modifications.