KAFKA-6927: Message down-conversion causes Out Of Memory on broker #4871

dhruvilshah3 · 2018-04-13T20:38:21Z

Implementation for lazy down-conversion in a chunked manner for efficient memory usage during down-conversion. This pull request is mainly to get initial feedback on the direction of the patch. The patch includes all the main components from KIP-283.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…orking.

dhruvilshah3 · 2018-05-08T16:49:35Z

Retest this please

dhruvilshah3 · 2018-05-08T17:04:46Z

@hachikuji could you please take a look. Thanks!

hachikuji · 2018-05-30T00:01:31Z

clients/src/main/java/org/apache/kafka/common/protocol/types/Type.java

@@ -556,7 +557,7 @@ public void write(ByteBuffer buffer, Object o) {
        }

        @Override
-        public Records read(ByteBuffer buffer) {
+        public MemoryRecords read(ByteBuffer buffer) {


We have a check above in write for FileRecords. Now that we have additional types to worry about, maybe we should change that check to to !(o instanceof MemoryRecords)?

hachikuji · 2018-05-30T00:08:54Z

core/src/main/scala/kafka/server/KafkaApis.scala

+          // as possible. With KIP-283, we have the ability to lazily down-convert in a chunked manner. The lazy, chunked
+          // down-conversion always guarantees that at least one batch of messages is down-converted and sent out to the
+          // client.
+          val converted = new LazyDownConversionRecords(tp, unconvertedFetchResponse.records, magic, fetchContext.getFetchOffset(tp).get, Time.SYSTEM)


We can use the time field instead of Time.SYSTEM

hachikuji · 2018-05-30T00:20:33Z

core/src/main/scala/kafka/server/KafkaApis.scala

+            FetchResponse.INVALID_LAST_STABLE_OFFSET, unconvertedFetchResponse.logStartOffset, unconvertedFetchResponse.abortedTransactions,
+            converted)
+        }
+      }.getOrElse(new FetchResponse.PartitionData[BaseRecords](unconvertedFetchResponse.error, unconvertedFetchResponse.highWatermark,


Well, if we have to construct a new instance anyway, I'd suggest simplifying this inner def a little bit. We can avoid the PartitionData object and pass the Records instance directly. Then we can construct the new PartitionData in the caller.

hachikuji · 2018-05-30T00:25:52Z

core/src/main/scala/kafka/server/KafkaApis.scala

-  private def sendResponseMaybeThrottle(request: RequestChannel.Request, createResponse: Int => AbstractResponse): Unit = {
+  private def sendResponseMaybeThrottle(request: RequestChannel.Request,
+                                        createResponse: Int => AbstractResponse,
+                                        processingStatsCallback: Option[FetchResponseStats => Unit] = None): Unit = {


It is a little odd to have a generic sendResponseMaybeThrottle method which has a callback which only makes sense when sending a fetch response. Maybe we should make the mechanism more generic? For example, maybe it can be an onComplete callback which gives access to the Send object?

hachikuji · 2018-05-30T00:26:47Z

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -16,6 +16,7 @@
 */
 package kafka.server

+


nit: probably unintentional

hachikuji · 2018-05-30T00:28:30Z

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

 import kafka.common.TopicAndPartition
 import kafka.log.LogConfig
 import kafka.network.SocketServer
 import kafka.security.auth._
 import kafka.server.{BaseRequestTest, KafkaConfig}
 import kafka.utils.TestUtils
 import org.apache.kafka.clients.admin.NewPartitions
-import org.apache.kafka.clients.consumer.OffsetAndMetadata
+import org.apache.kafka.clients.consumer.{OffsetAndMetadata, _}


nit: we can get rid of OffsetAndMetadata in this import

hachikuji

Thanks, left a few more comments. LGTM overall.

hachikuji · 2018-05-30T05:54:47Z

core/src/main/scala/kafka/server/KafkaApis.scala

-        def processingStatsCallback(recordConversionStats: FetchResponseStats): Unit = {
-          recordConversionStats.foreach { case (tp, info) =>
-            updateRecordsProcessingStats(request, tp, info)
+        def onComplete(send: Send): Unit = {


nit: we could name this something more descriptive like updateConversionStats.

hachikuji · 2018-05-30T05:56:33Z

clients/src/main/java/org/apache/kafka/common/record/LazyDownConversionRecords.java

+    /**
+     * {@inheritDoc}
+     * Note that we do not have a way to return the exact size of down-converted messages, so we return the size of the
+     * pre-down-converted messages. The consumer however expects at least one full batch of messages to be sent out so


This comment is out of place after we moved the logic into the constructor.

hachikuji · 2018-05-30T06:01:07Z

clients/src/main/java/org/apache/kafka/common/record/RecordConversionStats.java


-    public static final RecordsProcessingStats EMPTY = new RecordsProcessingStats(0L, 0, -1);
+    public static final RecordConversionStats EMPTY = new RecordConversionStats(0L, 0, -1);


Not from this patch, but it seems more natural to use 0 for conversionTimeNanos here. If we didn't convert anything, than I'd expect it took no time 😉.

Also, maybe we can add a no-arg constructor which initializes everything to 0. We have this in at least one other location.

hachikuji · 2018-05-30T06:04:14Z

clients/src/test/java/org/apache/kafka/common/record/LazyDownConversionRecordsTest.java

+    }
+
+    private void doTestConversion(CompressionType compressionType, byte toMagic, DownConversionTest test) throws IOException {
+        System.out.println("{" + compressionType + ", " + toMagic + ", " + test + "}");


Can we remove this? We typically don't have printlns in test code.

If a particular test case is flaky and fails occasionally, wouldn't this give us more information about which particular case is being flaky?

With the parameterized test, we don't require this println so I removed it.

hachikuji · 2018-05-30T06:05:58Z

clients/src/test/java/org/apache/kafka/common/record/LazyDownConversionRecordsTest.java

+
+    @Test
+    public void testConversion() throws IOException {
+        for (byte toMagic = RecordBatch.MAGIC_VALUE_V0; toMagic <= RecordBatch.CURRENT_MAGIC_VALUE; toMagic++) {


Another way to do this is using a parameterized test case. You can see MemoryRecordsBuilderTest for an example how to do this.

hachikuji · 2018-05-30T06:19:51Z

core/src/main/scala/kafka/network/RequestChannel.scala

-    request.responseCompleteTimeNanos = Time.SYSTEM.nanoseconds
-    if (request.apiLocalCompleteTimeNanos == -1L) request.apiLocalCompleteTimeNanos = Time.SYSTEM.nanoseconds
+  abstract class Response(val request: Request) {
+    val nowNs = Time.SYSTEM.nanoseconds


We could probably do this in a locally block since we don't need nowNs except to initialize the fields below.

hachikuji · 2018-05-30T06:21:05Z

core/src/main/scala/kafka/network/RequestChannel.scala

-        val responseString =
-          if (response.responseSend.isDefined)
-            response.responseAsString.getOrElse(
+        val responseString = response match {


Can we give Response a responseString method and skip this matching? The default implementation can return an empty string.

hachikuji · 2018-05-30T06:22:47Z

core/src/main/scala/kafka/network/SocketServer.scala

+        updateRequestMetrics(response)
+
+        // Invoke send completion callback
+        response match {


Similar to the comment above. If you find yourself matching on types, it is worth considering whether we can instead add a method to the type. Here we could have Response.onComplete, and the default implementation can do nothing.

hachikuji · 2018-05-30T06:23:56Z

core/src/main/scala/kafka/server/KafkaApis.scala

-import org.apache.kafka.common.utils.{Time, Utils}
-import org.apache.kafka.common.{Node, TopicPartition}
-import org.apache.kafka.common.requests.{SaslAuthenticateResponse, SaslHandshakeResponse}
+import org.apache.kafka.common.requests.{SaslAuthenticateResponse, SaslHandshakeResponse, Resource => RResource, ResourceType => RResourceType, _}


nit: the Sasl* imports are redundant given the wildcard. I sort of hate all the flexibility that scala gives for importing things.

hachikuji · 2018-05-30T06:27:12Z

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -471,7 +470,7 @@ class ReplicaManager(val config: KafkaConfig,
                    entriesPerPartition: Map[TopicPartition, MemoryRecords],
                    responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
                    delayedProduceLock: Option[Lock] = None,
-                    processingStatsCallback: Map[TopicPartition, RecordsProcessingStats] => Unit = _ => ()) {
+                    processingStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => ()) {


nit: one more rename

hachikuji · 2018-05-30T23:12:11Z

core/src/main/scala/kafka/network/RequestChannel.scala

+    override def onComplete: Option[Send => Unit] = onCompleteCallback
+
+    override def toString: String =
+      s"Response(responseType=SendResponse, request=$request, responseSend=$responseSend), responseAsString=$responseAsString"


nit: there are a lot of "responses" here. Maybe we simplify a little:

s"Response(type=Send, request=$request, send=$responseSend), asString=$responseAsString"

Also, it looks like we have changed the location of the end parenthesis, was that intentional?

hachikuji · 2018-05-30T23:14:45Z

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -2255,23 +2271,25 @@ class KafkaApis(val requestChannel: RequestChannel,
      sendErrorResponseExemptThrottle(request, e)
  }

-  // Throttle the channel if the request quota is enabled but has been violated. Regardless of throttling, send the


We seem to have lost this comment.

hachikuji · 2018-05-30T23:26:18Z

core/src/main/scala/kafka/network/RequestChannel.scala

-              throw new IllegalStateException("responseAsString should always be defined if request logging is enabled"))
-          else ""
-
+        val responseString = response.responseString.getOrElse("responseAsString should always be defined if request logging is enabled")


Should we be throwing IllegalStateException?

hachikuji · 2018-05-30T23:30:03Z

core/src/main/scala/kafka/server/ThrottledChannel.scala

  * @param channelThrottlingCallback Callback for channel throttling
  */
-class ThrottledChannel(val time: Time, val throttleTimeMs: Int, channelThrottlingCallback: (ResponseAction) => Unit)
+class ThrottledChannel(val request: RequestChannel.Request, val time: Time, val throttleTimeMs: Int, channelThrottlingCallback: (Response) => Unit)


nit: parenthesis are unnecessary

hachikuji · 2018-05-30T23:36:48Z

core/src/main/scala/kafka/network/SocketServer.scala

@@ -730,7 +736,8 @@ private[kafka] class Processor(val id: Int,
    }
  }

-  private def updateRequestMetrics(response: RequestChannel.Response) {
+  private def updateRequestMetrics(response: RequestChannel.Response): Unit = {
+    // Record the amount of time this request spent on the network thread


Was this comment added intentionally? The request metrics include more than how much time was spent on the network thread.

hachikuji

Thanks, LGTM. Just a few additional comments.

hachikuji · 2018-05-30T23:42:27Z

core/src/main/scala/kafka/server/ClientQuotaManager.scala

-
-  def throttle(session: Session, clientId: String, throttleTimeMs: Int,
-               channelThrottlingCallback: (ResponseAction) => Unit) {
+  def throttle(request: RequestChannel.Request, throttleTimeMs: Int, channelThrottlingCallback: (Response) => Unit): Unit = {


nit: unneeed parenthesis

hachikuji · 2018-05-30T23:50:14Z

core/src/main/scala/kafka/network/RequestChannel.scala

-                 val responseAsString: Option[String]) {
-    request.responseCompleteTimeNanos = Time.SYSTEM.nanoseconds
-    if (request.apiLocalCompleteTimeNanos == -1L) request.apiLocalCompleteTimeNanos = Time.SYSTEM.nanoseconds
+  abstract class Response(val request: Request) {


It seems more appropriate to treat this as a ResponseAction instead of a Response itself, especially given the new throttling behavior.

I will consider taking this up in a follow-on patch.

hachikuji · 2018-05-31T00:20:25Z

retest this please

ijuma · 2018-05-31T07:34:29Z

clients/src/main/java/org/apache/kafka/common/record/LazyDownConversionRecordsSend.java

+                //      Length => Int32
+                //      ...
+                // TODO: check if there is a better way to encapsulate this logic, perhaps in DefaultRecordBatch
+                log.debug("Constructing fake message batch for topic-partition {" + topicPartition() + "} for remaining length " + remaining);


We should not do string concatenation like this. Java only has strict evaluation so we are doing this operation even if the log is not enabled. Please use slf4j's interpolation. Also, no TODOs please.

I will address this in a follow-on PR

omkreddy · 2018-06-07T12:59:05Z

@hachikuji @dhruvilshah3
I am observing below exceptions while running old consumer related code (kafka.tools.TestLogCleaning) on compacted topics. I am not observing on normal topics. just want to let you guys know.

Server log:

[2018-06-07 18:15:58,164] WARN [SocketServer brokerId=0] Unexpected error from /10.200.5.75; closing connection (org.apache.kafka.common.network.Selector)
java.util.NoSuchElementException
	at org.apache.kafka.common.utils.AbstractIterator.next(AbstractIterator.java:52)
	at org.apache.kafka.common.record.LazyDownConversionRecordsSend.writeTo(LazyDownConversionRecordsSend.java:59)
	at org.apache.kafka.common.record.RecordsSend.writeTo(RecordsSend.java:58)
	at org.apache.kafka.common.record.MultiRecordsSend.writeTo(MultiRecordsSend.java:93)
	at org.apache.kafka.common.network.KafkaChannel.send(KafkaChannel.java:339)
	at org.apache.kafka.common.network.KafkaChannel.write(KafkaChannel.java:310)
	at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:509)
	at org.apache.kafka.common.network.Selector.poll(Selector.java:424)
	at kafka.network.Processor.poll(SocketServer.scala:680)
	at kafka.network.Processor.run(SocketServer.scala:585)
	at java.base/java.lang.Thread.run(Thread.java:844)

client reconnect failing repeatedly with ClosedChannelException:

[2018-06-07 18:15:39,176] INFO Reconnect due to error: (kafka.consumer.SimpleConsumer:68)
java.nio.channels.ClosedChannelException
	at kafka.network.BlockingChannel.send(BlockingChannel.scala:112)
	at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:88)
	at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:86)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:135)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:135)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:135)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:31)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:134)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:134)
	at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:134)
	at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:31)
	at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:133)
	at kafka.consumer.ConsumerFetcherThread.fetch(ConsumerFetcherThread.scala:122)
	at kafka.consumer.ConsumerFetcherThread.fetch(ConsumerFetcherThread.scala:37)
	at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:149)
	at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:114)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:82)
[2018-06-07 18:15:39,176] DEBUG Disconnecting from 10.200.5.75:9092 (kafka.consumer.SimpleConsumer:62)

dhruvilshah3 · 2018-06-07T14:55:56Z

@omkreddy Thanks for reporting this. I will email you to get some more information about the issue.

ijuma · 2018-06-07T15:25:02Z

@omkreddy I suggest filing a JIRA with more details.

omkreddy · 2018-06-07T17:09:40Z

Provided some more details to Dhruvil. Requested him to raise a JIRA, If he thinks its a issue.

…n broker [KIP-283] (apache#4871) Implementation for lazy down-conversion in a chunked manner for efficient memory usage during down-conversion. This pull request is mainly to get initial feedback on the direction of the patch. The patch includes all the main components from KIP-283. Reviewers: Jason Gustafson <jason@confluent.io>

dhruvilshah3 added 3 commits April 11, 2018 13:59

LazyDownConversionRecords implementation.

2185e03

Remove topic configuration changes.

19c30f5

Maintain metrics for lazy down-conversion.

e619124

dhruvilshah3 changed the title ~~KAFKA 6709: Message down-conversion causes Out Of Memory on broker~~ KAFKA-6709: Message down-conversion causes Out Of Memory on broker Apr 13, 2018

Code cleanup.

4b04c09

dhruvilshah3 force-pushed the KAFKA-6709 branch from 8154268 to 4b04c09 Compare April 14, 2018 05:32

dhruvilshah3 added 2 commits April 23, 2018 21:43

Down-conversion unit tests

1226496

Use SerializableRecords for LazyDownConversionRecords. Verified w…

2b4e36f

…orking.

dhruvilshah3 changed the title ~~KAFKA-6709: Message down-conversion causes Out Of Memory on broker~~ [WIP] KAFKA-6709: Message down-conversion causes Out Of Memory on broker Apr 27, 2018

dhruvilshah3 added 6 commits April 27, 2018 16:59

LazyDownConversionRecords implementation.

90e87bc

Remove topic configuration changes.

ca7c623

Maintain metrics for lazy down-conversion.

3dcb201

Code cleanup.

d88fd93

Down-conversion unit tests

910525a

Use SerializableRecords for LazyDownConversionRecords. Verified w…

53a76dd

…orking.

dhruvilshah3 force-pushed the KAFKA-6709 branch from 2b4e36f to 53a76dd Compare April 28, 2018 00:03

dhruvilshah3 added 6 commits April 27, 2018 17:12

Rebase against trunk.

4ea1deb

Minor refactoring and checkStyle changes.

46c067e

Added logging to LazyDownConversionRecords.

69cee5f

Unit test

ad7daf4

Always down-convert first batch.

6d44315

Added some comments.

1289818

dhruvilshah3 force-pushed the KAFKA-6709 branch from 43748b9 to 1289818 Compare May 3, 2018 20:38

dhruvilshah3 changed the title ~~[WIP] KAFKA-6709: Message down-conversion causes Out Of Memory on broker~~ KAFKA-6709: Message down-conversion causes Out Of Memory on broker May 3, 2018

checkstyle

a0474a8

dhruvilshah3 force-pushed the KAFKA-6709 branch from 611409c to a0474a8 Compare May 3, 2018 21:46

dhruvilshah3 added 2 commits May 8, 2018 11:09

Fix failing unit tests.

b31b6ec

Fix FetchRequestTest

8d6b5b9

data --> unconvertedFetchResponse

cc03a52

hachikuji reviewed May 30, 2018

View reviewed changes

Address review comments

fed927f

hachikuji approved these changes May 30, 2018

View reviewed changes

dhruvilshah3 added 4 commits May 30, 2018 08:59

Address review comments.

002b5e5

checkstyle

af74c13

Merge branch 'trunk' into KAFKA-6709

74f714d

Fix ThrottledChannelExpirationTest

9deb553

hachikuji reviewed May 30, 2018

View reviewed changes

Address review comments

2e35d5c

dhruvilshah3 force-pushed the KAFKA-6709 branch from ca7ee00 to 2e35d5c Compare May 30, 2018 23:57

hachikuji merged commit 837f31d into apache:trunk May 31, 2018

ijuma reviewed May 31, 2018

View reviewed changes


		public static final RecordsProcessingStats EMPTY = new RecordsProcessingStats(0L, 0, -1);
		public static final RecordConversionStats EMPTY = new RecordConversionStats(0L, 0, -1);

KAFKA-6927: Message down-conversion causes Out Of Memory on broker #4871

KAFKA-6927: Message down-conversion causes Out Of Memory on broker #4871

Conversation

dhruvilshah3 commented Apr 13, 2018 • edited

Committer Checklist (excluded from commit message)

dhruvilshah3 commented May 8, 2018

dhruvilshah3 commented May 8, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji commented May 31, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omkreddy commented Jun 7, 2018 • edited

dhruvilshah3 commented Jun 7, 2018

ijuma commented Jun 7, 2018

omkreddy commented Jun 7, 2018 • edited

dhruvilshah3 commented Apr 13, 2018 •

edited

omkreddy commented Jun 7, 2018 •

edited

omkreddy commented Jun 7, 2018 •

edited