It's valid to abort a txn that hasn't been started

There may be cases where dependant code starts a transaction using the
producer's `#transaction` block syntax and then an exception is raised
within that block *before* the transaction manager actually opens up a
transaction. In such cases, prior to this change, the real exception is
obfuscated, because the transaction manager would raise an exception
about not being in a valid transactional state.

With this change, attempting to abort a transaction that has not
actually been opened will no longer result in an exception. Instead it
will log a warning message and leave the transaction manager in its
ready state.

Author: jwilger

lib/kafka/transaction_manager.rb

@@ -192,7 +192,8 @@ def abort_transaction
       end
 
       unless @transaction_state.in_transaction?
-        raise Kafka::InvalidTxnStateError, 'Transaction is not valid to abort'
+        @logger.warn('Aborting transaction that was never opened on brokers')
+        return
       end
 
       @transaction_state.transition_to!(TransactionStateMachine::ABORTING_TRANSACTION)
@@ -250,6 +251,10 @@ def error?
       @transaction_state.error?
     end
 
+    def ready?
+      @transaction_state.ready?
+    end
+
     def close
       if in_transaction?
         @logger.warn("Aborting pending transaction ...")

spec/transaction_manager_spec.rb

@@ -401,17 +401,15 @@
         manager.init_transactions
       end
 
-      it 'raises exception' do
-        expect do
-          manager.abort_transaction
-        end.to raise_error(Kafka::InvalidTxnStateError, /transaction is not valid to abort/i)
+      it 'does not raise an exception' do
+        expect { manager.abort_transaction }.not_to raise_error
       end
 
-      it 'changes state to error' do
+      it 'leaves transaction manager in ready state' do
         begin
           manager.abort_transaction
         rescue; end
-        expect(manager.error?).to eql(true)
+        expect(manager.ready?).to eql(true)
       end
     end