@Test public void testMaybeAddPartitionToTransaction() { long pid = 13131L; short epoch = 1; TopicPartition partition = new TopicPartition("foo", 0); doInitTransactions(pid, epoch); transactionManager.beginTransaction(); transactionManager.maybeAddPartitionToTransaction(partition); assertTrue(transactionManager.hasPartitionsToAdd()); assertFalse(transactionManager.isPartitionAdded(partition)); assertTrue(transactionManager.isPartitionPendingAdd(partition)); prepareAddPartitionsToTxn(partition, Errors.NONE); sender.run(time.milliseconds()); assertFalse(transactionManager.hasPartitionsToAdd()); assertTrue(transactionManager.isPartitionAdded(partition)); assertFalse(transactionManager.isPartitionPendingAdd(partition)); transactionManager.maybeAddPartitionToTransaction(partition); assertFalse(transactionManager.hasPartitionsToAdd()); assertTrue(transactionManager.isPartitionAdded(partition)); assertFalse(transactionManager.isPartitionPendingAdd(partition)); }
public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) { failIfNotReadyForSend(); if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition)) return; log.debug("{}Begin adding new partition {} to transaction", logPrefix, topicPartition); newPartitionsInTransaction.add(topicPartition); }
TransactionManager { public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) { failIfNotReadyForSend(); if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition)) return; log.debug("{}Begin adding new partition {} to transaction", logPrefix, topicPartition); newPartitionsInTransaction.add(topicPartition); } }
TransactionManager { public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) { failIfNotReadyForSend(); if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition)) return; log.debug("{}Begin adding new partition {} to transaction", logPrefix, topicPartition); newPartitionsInTransaction.add(topicPartition); } TransactionManager(String transactionalId, int transactionTimeoutMs, long retryBackoffMs);   TransactionManager(); }
TransactionManager { public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) { failIfNotReadyForSend(); if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition)) return; log.debug("{}Begin adding new partition {} to transaction", logPrefix, topicPartition); newPartitionsInTransaction.add(topicPartition); } 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(); }
TransactionManager { public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) { failIfNotReadyForSend(); if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition)) return; log.debug("{}Begin adding new partition {} to transaction", logPrefix, topicPartition); newPartitionsInTransaction.add(topicPartition); } 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; }