@ Override public void close ( ) throws IOException { try { if ( consumer != null && consumer instanceof Closeable ) { // Call close in a new transaction. // TODO (terence): Actually need to coordinates with other flowlets to drain the queue. TransactionContext txContext = dataFabricFacade . createTransactionManager ( ) ; txContext . start ( ) ; try { ( ( Closeable ) consumer ) . close ( ) ; txContext . finish ( ) ; } catch ( TransactionFailureException e ) { LOG . warn ( "Fail to commit transaction when closing consumer." ) ; txContext . abort ( ) ; } } } catch ( Exception e ) { LOG . warn ( "Fail to close queue consumer." , e ) ; } consumer = null ; }
Close the current consumer if there is one .