### Question:
TransactionManager { public synchronized void failIfNotReadyForSend() { if (hasError()) throw new KafkaException("Cannot perform send because at least one previous transactional or " + "idempotent request has failed with errors.", lastError); if (isTransactional()) { if (!hasProducerId()) throw new IllegalStateException("Cannot perform a 'send' before completing a call to initTransactions " + "when transactions are enabled."); if (currentState != State.IN_TRANSACTION) throw new IllegalStateException("Cannot call send in state " + currentState); } } TransactionManager(String transactionalId, int transactionTimeoutMs, long retryBackoffMs);   TransactionManager(); synchronized TransactionalRequestResult initializeTransactions(); synchronized void beginTransaction(); synchronized TransactionalRequestResult beginCommit(); synchronized TransactionalRequestResult beginAbort(); synchronized TransactionalRequestResult sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                                                            String consumerGroupId); synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition); synchronized void failIfNotReadyForSend(); String transactionalId(); boolean hasProducerId(); boolean isTransactional(); final String logPrefix; }### Answer:
@Test(expected = IllegalStateException.class) public void testFailIfNotReadyForSendNoProducerId() { transactionManager.failIfNotReadyForSend(); }
@Test public void testFailIfNotReadyForSendIdempotentProducer() { TransactionManager idempotentTransactionManager = new TransactionManager(); idempotentTransactionManager.failIfNotReadyForSend(); }
@Test(expected = IllegalStateException.class) public void testFailIfNotReadyForSendNoOngoingTransaction() { long pid = 13131L; short epoch = 1; doInitTransactions(pid, epoch); transactionManager.failIfNotReadyForSend(); }