Skip to content

KAFKA-6768; Transactional producer may hang in close with pending requests #4842

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 1 commit into from
Apr 9, 2018
+21 −1
Merged

KAFKA-6768; Transactional producer may hang in close with pending requests #4842

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,7 +329,7 @@ private boolean maybeSendTransactionalRequest(long now) {
return false;

AbstractRequest.Builder<?> requestBuilder = nextRequestHandler.requestBuilder();
while (running) {
while (!forceClose) {

@hachikuji hachikuji Apr 9, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The problem was basically that initiateClose set running to true, which prevented the transactional request from being sent. That in turn kept batched records stuck in the accumulator and prevented shutdown from completing.

@apurvam apurvam Apr 9, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit (for documentation): initiateClose would set running to false.

Node targetNode = null;
try {
if (nextRequestHandler.needsCoordinator()) {
Expand Down
20 changes: 20 additions & 0 deletions ...nts/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,26 @@ public void setup() {
client.setNode(brokerNode);
}

@Test
public void testSenderShutdownWithPendingAddPartitions() throws Exception {
long pid = 13131L;
short epoch = 1;
doInitTransactions(pid, epoch);
transactionManager.beginTransaction();

transactionManager.maybeAddPartitionToTransaction(tp0);
FutureRecordMetadata sendFuture = accumulator.append(tp0, time.milliseconds(), "key".getBytes(),
"value".getBytes(), Record.EMPTY_HEADERS, null, MAX_BLOCK_TIMEOUT).future;

prepareAddPartitionsToTxn(tp0, Errors.NONE);
prepareProduceResponse(Errors.NONE, pid, epoch);

sender.initiateClose();
sender.run();

assertTrue(sendFuture.isDone());
}

@Test
public void testEndTxnNotSentIfIncompleteBatches() {
long pid = 13131L;
Expand Down