title: org.apache.kafka.common.errors.RecordTooLargeException in Flume Kafka Sink
body: <p>I am trying to read data from JMS source and pushing them into KAFKA topic, while doing that after few hours i observed that pushing frequency to the KAFKA topic became almost zero and after some initial analysis i found following exception in FLUME logs .</p>
  
  <pre><code>28 Feb 2017 16:35:44,758 ERROR [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.SinkRunner$PollingRunner.run:158)  - Unable to deliver event. Exception follows.
  org.apache.flume.EventDeliveryException: Failed to publish events
          at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:252)
          at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67)
          at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145)
          at java.lang.Thread.run(Thread.java:745)
  Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.RecordTooLargeException: The message is 1399305 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
          at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.&lt;init&gt;(KafkaProducer.java:686)
          at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:449)
          at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:212)
          ... 3 more
  Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 1399305 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
  </code></pre>
  
  <p>my flume shows the current set value (in logs ) for  max.request.size as 1048576 , which is clearly very less than  1399305 , increasing this max.request.size may eliminate these exception but am unable to find correct place for updating that value .</p>
  
  <p>My flume.config , </p>
  
  <pre><code>a1.sources = r1
  a1.channels = c1
  a1.sinks = k1
  
  a1.channels.c1.type = file
  a1.channels.c1.transactionCapacity = 1000
  a1.channels.c1.capacity = 100000000
  a1.channels.c1.checkpointDir = /data/flume/apache-flume-1.7.0-bin/checkpoint
  a1.channels.c1.dataDirs = /data/flume/apache-flume-1.7.0-bin/data
  
  a1.sources.r1.type = jms
  
  a1.sources.r1.interceptors.i1.type = timestamp
  a1.sources.r1.interceptors.i1.preserveExisting = true
  
  a1.sources.r1.channels = c1
  a1.sources.r1.initialContextFactory = some context urls
  a1.sources.r1.connectionFactory = some_queue
  a1.sources.r1.providerURL = some_url 
  #a1.sources.r1.providerURL = some_url
  a1.sources.r1.destinationType = QUEUE
  a1.sources.r1.destinationName = some_queue_name 
  a1.sources.r1.userName = some_user
  a1.sources.r1.passwordFile= passwd
  
  a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
  a1.sinks.k1.kafka.topic = some_kafka_topic
  a1.sinks.k1.kafka.bootstrap.servers = some_URL
  a1.sinks.k1.kafka.producer.acks = 1
  a1.sinks.k1.flumeBatchSize = 1
  a1.sinks.k1.channel = c1
  </code></pre>
  
  <p>Any help will be really appreciated !!</p>