KAFKA-5239: Producer buffer pool allocates memory inside a lock

[smccauliff]

Move byte buffer allocation out of lock.
Add unit test for restoring count when OOM is thrown from byte buffer allocation.

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3898/
Test FAILed (JDK 8 and Scala 2.12).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3909/
Test FAILed (JDK 7 and Scala 2.11).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3919/
Test PASSed (JDK 8 and Scala 2.12).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3930/
Test PASSed (JDK 7 and Scala 2.11).

[becketqin]

If we change the definition of the available memory here, we may have to change a few other places as well. e.g. deallocate(), unallocatedMemory(), etc.

[smccauliff]

I added the original comment; it is not correct.

[smccauliff]

Lines 234-236 confused me. But I see that freeUp() will potentially add the count of bytes back into availableMemory.

[becketqin]

Thanks for the explanation. It seems the comment were still not updated? Have you pushed the change?

[smccauliff]

/** Poolable + non-poolable memory. */ is the correct comment. No?

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4934/
Test PASSed (JDK 7 and Scala 2.11).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4918/
Test PASSed (JDK 8 and Scala 2.12).

[becketqin]

Thanks for the patch. LGTM.

[ijuma]

Not sure if it's related, but the following test failed in Jenkins:

java.lang.AssertionError: available memory8
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.assertTrue(Assert.java:41)
	at org.apache.kafka.clients.producer.internals.BufferPoolTest.testBlockTimeout(BufferPoolTest.java:191)

https://builds.apache.org/blue/organizations/jenkins/kafka-trunk-jdk8/detail/kafka-trunk-jdk8/1660/tests

[becketqin]

@ijuma I ran the tests a couple of times locally and cannot reproduce the issue. It seems an intermittent failure related to the delayedDeallocate. It does not look related to this patch.