KAFKA-3718: propagate all KafkaConfig __consumer_offsets configs to OffsetConfig instantiation

[onurkaraman]

Kafka has two configurable compression codecs: the one used by the client (source codec) and the one finally used when storing into the log (target codec). The target codec defaults to KafkaConfig.compressionType and can be dynamically configured through zookeeper.

The GroupCoordinator appends group membership information into the __consumer_offsets topic by:

1. making a message with group membership information
2. making a MessageSet with the single message compressed with the source codec
3. doing a log.append on the MessageSet

Without this patch, KafkaConfig.offsetsTopicCompressionCodec doesn't get propagated to OffsetConfig instantiation, so GroupMetadataManager uses a source codec of NoCompressionCodec when making the MessageSet. Let's say we have enough group information such that the message formed exceeds KafkaConfig.messageMaxBytes before compression but would fall below the threshold after compression using our source codec. Even if we had dynamically configured __consumer_offsets with our favorite compression codec, the log.append will throw RecordTooLargeException during analyzeAndValidateMessageSet since the message was unexpectedly uncompressed instead of having been compressed with the source codec defined by KafkaConfig.offsetsTopicCompressionCodec.

[omkreddy]

sometime back I submitted similar patch for old ConsumerCoordinator (now replaced with GroupCoordinator)
https://issues.apache.org/jira/browse/KAFKA-2159. I will close KAFKA-2159 JIRA.

[omkreddy]

Should we use offsetConfig.offsetsTopicCompressionCodec here also?
https://github.com/onurkaraman/kafka/blob/d65726de2e4a196837de7bb2e11f397482826dca/core/src/main/scala/kafka/coordinator/GroupCoordinator.scala#L68

[ijuma]

Thanks for the PR. Can we have a test for this? @hachikuji, can you please take a look?

[onurkaraman]

@ijuma not really sure how you want me to test this.

I mean if we really wanted to, I guess I can unit test the GroupCoordinator constructor by instantiating a GroupCoordinator with a custom KafkaConfig.offsetsTopicCompressionCodec defined and verify that groupCoordinator.offsetConfig.offsetsTopicCompressionCodec has picked up the modification? Testing constructors seems kinda lame though.

[onurkaraman]

@omkreddy I think the default compression.type (target codec) is "producer", meaning it adopts whatever the codec of the MessageSet was coming in. So it seems that updating the property in GroupCoordinator. offsetsTopicConfigs isn't really required. Thoughts?

Pinging @hachikuji for review.

[hachikuji]

@onurkaraman Good find. I guess the only difference if we set compression.type is how we'd react to a future config change. For example, if we initially set offsets.topic.compression.codec to snappy and write that to Zk as the topic's compression.type, then if we later change offsets.topic.compression.codec to gzip, we'd first compress as gzip and then recompress as snappy on writing (unless we change the code to alter the topic configuration). Alternatively, with the change as it is, we'll leave the data compressed according to whatever the current configuration is. That seems preferable to me.

But actually I'm a little confused. Isn't the compression.type for this topic explicitly set to uncompressed on that line above that @omkreddy points to above? Wouldn't that mean that we'd decompress before writing even with this patch? Seems like we actually need to change that to "producer." Maybe I'm missing something?

[omkreddy]

@hachikuji Yes, we should set "producer" as compression.type.

props.put(LogConfig.CompressionTypeProp, ProducerCompressionCodec.name)

@onurkaraman as mentioned by @hachikuji, for this topic we are overriding the target.codec to uncompressed. We should change it to producer (or) we should remove the line and change below default OffsetConfig codec to "producer"
https://github.com/onurkaraman/kafka/blob/d65726de2e4a196837de7bb2e11f397482826dca/core/src/main/scala/kafka/coordinator/OffsetConfig.scala#L58

[onurkaraman]

Yeah my bad. My previous comment wasn't right. Unless the config is overridden, compression.type is producer. But we explicitly set the compression.type for the __consumer_offsets topic in the line @omkreddy stated.

It looks like compression.type can only be configured on a per-topic level during topic creation through AdminUtils.createTopic or through dynamic config with kafka-configs.sh. I had been doing dynamic config.

[onurkaraman]

I force pushed an update that explicitly sets the compression.type to producer so that the target codec always just adopts the source codec.

This seems better than removing the property altogether since that would make the target codec whatever the compression.type is at the time, which may not be producer. It would probably be surprising to set offsetsTopicCompressionCodec only to see the final target codec be something else.

[ijuma]

@onurkaraman, with regards to testing, I think it might be useful to have an integration test where we configure the offsets topic to be compressed and then we verify that the topic is compressed with the expected compression algorithm.

[omkreddy]

LGTM

[hachikuji]

LGTM. An integration test case would be nice if it's not too much trouble. Maybe we could commit an offset and then read it back using the simple consumer and verify that the right codec was used?

[onurkaraman]

@ijuma I force pushed an update that adds an integration test checking that the offsetsTopicCompressionCodec has been propagated.

[onurkaraman]

@ijuma any chance of checking this in today now that 0.10.0.0 is released?

[ijuma]

@onurkaraman, I'll try, but the day is almost over in London.

[onurkaraman]

Right. Timezones. Sorry I forgot.

[ijuma]

Thanks for adding the test. I made a few changes to make it a bit more readable and to improve the error if the test fails:

package integration.kafka.api

import kafka.common.TopicAndPartition
import kafka.integration.KafkaServerTestHarness
import kafka.log.Log
import kafka.message.GZIPCompressionCodec
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.OffsetAndMetadata
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.internals.TopicConstants
import org.apache.kafka.common.protocol.SecurityProtocol
import org.junit.Test
import org.junit.Assert._

import scala.collection.JavaConverters._
import java.util.Properties

class GroupCoordinatorIntegrationTest extends KafkaServerTestHarness {
  val offsetsTopicCompressionCodec = GZIPCompressionCodec
  val overridingProps = new Properties()
  overridingProps.put(KafkaConfig.OffsetsTopicPartitionsProp, "1")
  overridingProps.put(KafkaConfig.OffsetsTopicCompressionCodecProp, offsetsTopicCompressionCodec.codec.toString)

  override def generateConfigs = TestUtils.createBrokerConfigs(1, zkConnect, enableControlledShutdown = false).map {
    KafkaConfig.fromProps(_, overridingProps)
  }

  @Test
  def testGroupCoordinatorPropagatesOfffsetsTopicCompressionCodec() {
    val consumer = TestUtils.createNewConsumer(TestUtils.getBrokerListStrFromServers(servers),
      securityProtocol = SecurityProtocol.PLAINTEXT)
    val offsetMap = Map(
      new TopicPartition(TopicConstants.GROUP_METADATA_TOPIC_NAME, 0) -> new OffsetAndMetadata(10, "")
    ).asJava
    consumer.commitSync(offsetMap)
    val logManager = servers.head.getLogManager

    def getGroupMetadataLogOpt: Option[Log] =
      logManager.getLog(TopicAndPartition(TopicConstants.GROUP_METADATA_TOPIC_NAME, 0))

    TestUtils.waitUntilTrue(() => getGroupMetadataLogOpt.exists(_.logSegments.exists(_.log.nonEmpty)),
      "Commit message not appended in time")

    val logSegments = getGroupMetadataLogOpt.get.logSegments
    val incorrectCompressionCodecs = logSegments.flatMap(_.log.map(_.message.compressionCodec)).filter(_ != offsetsTopicCompressionCodec)
    assertEquals("Incorrect compression codecs should be empty", Seq.empty, incorrectCompressionCodecs)

    consumer.close()
  }
}

What do you think?

[ijuma]

@onurkaraman, I left one suggestion. If you agree and integrate that, then I think we can merge this PR.

[onurkaraman]

Thanks for the cleanup @ijuma. Looks a lot cleaner. I force pushed the change.

[ijuma]

Thanks Onur, LGTM. Merging to trunk and 0.10.0.