Skip to content

KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced#1304

Closed
zhuchen1018 wants to merge 6 commits intoapache:trunkfrom
zhuchen1018:KAFKA-3648
Closed

KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced#1304
zhuchen1018 wants to merge 6 commits intoapache:trunkfrom
zhuchen1018:KAFKA-3648

Conversation

@zhuchen1018
Copy link
Copy Markdown
Contributor

@zhuchen1018 zhuchen1018 commented May 3, 2016 •
edited
Loading

maxTimeToBlock needs to be updated in each loop iteration. Also record waitTime before throwing TimeoutException

@lindong28
Copy link
Copy Markdown
Member

lindong28 commented May 3, 2016

Can you do maxTimeToBlock = Math.max(maxTimeToBlock - (endWait - startWait) / 1000000, 0)? Otherwise, LGTM.

@ijuma Can you take a look?

@ijuma
Copy link
Copy Markdown
Member

ijuma commented May 3, 2016

A minor suggestion, it would be better to use a new variable (maybe timeToBlock) instead of mutating the received method parameter.

It would be good to have a test for this.

@lindong28, do you know why we throw a TimeoutException without calling this.waitTime.record first?

@lindong28
Copy link
Copy Markdown
Member

lindong28 commented May 3, 2016

@ijuma No. It seems that we should record waitTime before TimeoutException is thrown.

@ijuma
Copy link
Copy Markdown
Member

ijuma commented May 3, 2016

Thanks for confirming. It would be good to fix that as part of this PR too.

@zhuchen1018
Copy link
Copy Markdown
Contributor Author

zhuchen1018 commented May 3, 2016

@ijuma The unit tests have been added and the two issues (a new variable & record waitTime before throwing TimeoutException) have been resolved. Can you take a look?

@lindong28
Copy link
Copy Markdown
Member

lindong28 commented May 3, 2016

LGTM

ijuma
ijuma reviewed May 3, 2016
View reviewed changes
Comment thread clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java Outdated
assertEquals("Allocation shouldn't have happened yet, waiting on memory.", 1L, allocation.getCount());
doDealloc.countDown(); // return the memory
allocation.await();
assertEquals("Allocation should succeed soon after de-allocation", true, allocation.await(1, TimeUnit.SECONDS));
Copy link
Copy Markdown
Member

@ijuma ijuma May 3, 2016

Choose a reason for hiding this comment

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

Nitpick: assertTrue can be used instead of assertEquals.

@ijuma
Copy link
Copy Markdown
Member

ijuma commented May 3, 2016

Thanks for the updates @zhuchen1018, I left a few more comments. Also, can you please update the PR title to:

KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced

And the PR description could be:

`maxTimeToBlock` needs to be updated in each loop iteration. Also record waitTime before throwing `TimeoutException`.

Thanks!

@zhuchen1018 zhuchen1018 changed the title KAFKA-3648: maxTimeToBlock should be enforced KAFKA-3648: maxTimeToBlock needs to be updated in each loop iteration. Also record waitTime before throwing TimeoutException. May 3, 2016
@zhuchen1018
Copy link
Copy Markdown
Contributor Author

zhuchen1018 commented May 3, 2016

@ijuma I have made changes as suggested. Can you please take another look? Thanks!

ijuma
ijuma reviewed May 3, 2016
View reviewed changes
Comment thread clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java Outdated
boolean waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);
long endWaitNs = time.nanoseconds();
long timeNs = Math.max(0L, endWaitNs - startWaitNs);
this.waitTime.record(timeNs, TimeUnit.NANOSECONDS.toMillis(endWaitNs));
Copy link
Copy Markdown
Member

@ijuma ijuma May 3, 2016

Choose a reason for hiding this comment

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

Sorry, I just realised that what I recommended here is wrong, the second parameter needs to be time.milliseconds().

Copy link
Copy Markdown
Contributor Author

@zhuchen1018 zhuchen1018 May 3, 2016

Choose a reason for hiding this comment

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

@ijuma Do you mean changing this.waitTime.record(timeNs, TimeUnit.NANOSECONDS.toMillis(endWaitNs)) to this.waitTime.record(timeNs, time.milliseconds())? I think the only difference between the two is the time to do long timeNs = Math.max(0L, endWaitNs - startWaitNs), which shouldn't take much time. Do we need this extra accuracy at the cost of one extra system call time.millionseconds()?

Copy link
Copy Markdown
Member

@ijuma ijuma May 3, 2016

Choose a reason for hiding this comment

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

Yes, that's what I mean. The issue is not the accuracy, but that System.nanoTime is not specified to return wall-clock time. So, unfortunately we need time.milliseconds (which is the same as System.currentTimeMillis). My original suggestion had the goal of saving the system call indeed, but that's not possible while maintaining the current semantics. Unfortunately, our Time interface is misleading in this respect (I filed https://issues.apache.org/jira/browse/KAFKA-2607 a while back to improve this).

Copy link
Copy Markdown
Contributor Author

@zhuchen1018 zhuchen1018 May 3, 2016

Choose a reason for hiding this comment

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

@ijuma Thanks! Fixed now.

@ijuma
Copy link
Copy Markdown
Member

ijuma commented May 3, 2016

Thanks @zhuchen1018, left two comments (one of them was regarding a mistake on my part).

With regards to the PR title and description, it seems like you updated the PR title with what should be the description (Git convention is that the PR title should be short and the description can be longer).

@zhuchen1018 zhuchen1018 changed the title KAFKA-3648: maxTimeToBlock needs to be updated in each loop iteration. Also record waitTime before throwing TimeoutException. KAFKA-3648: maxTimeToBlock needs to be updated in each loop iteration. Also record waitTime before throwing TimeoutException.KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced May 3, 2016
@zhuchen1018 zhuchen1018 changed the title KAFKA-3648: maxTimeToBlock needs to be updated in each loop iteration. Also record waitTime before throwing TimeoutException.KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced May 3, 2016
@asfgit asfgit closed this in 94e12a2 May 3, 2016
asfgit pushed a commit that referenced this pull request May 3, 2016
@zhuchen1018 @ijuma
KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced
138b37d 
 `maxTimeToBlock` needs to be updated in each loop iteration. Also record waitTime before throwing `TimeoutException`

Author: Chen Zhu <amandazhu19620701@gmail.com>

Reviewers: Dong Lin <lindong28@gmail.com>, Ismael Juma <ismael@juma.me.uk>

Closes #1304 from zhuchen1018/KAFKA-3648

(cherry picked from commit 94e12a2)
Signed-off-by: Ismael Juma <ismael@juma.me.uk>
@zhuchen1018 zhuchen1018 deleted the KAFKA-3648 branch May 3, 2016 23:24
gfodor pushed a commit to AltspaceVR/kafka that referenced this pull request Jun 3, 2016
@zhuchen1018 @gfodor
KAFKA-3648; maxTimeToBlock in BufferPool.allocate should be enforced
bc54fe7 
 `maxTimeToBlock` needs to be updated in each loop iteration. Also record waitTime before throwing `TimeoutException`

Author: Chen Zhu <amandazhu19620701@gmail.com>

Reviewers: Dong Lin <lindong28@gmail.com>, Ismael Juma <ismael@juma.me.uk>

Closes apache#1304 from zhuchen1018/KAFKA-3648
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@zhuchen1018 @lindong28 @ijuma