[
https://issues.apache.org/jira/browse/SAMZA-21?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13864846#comment-13864846
]
Jakob Homan commented on SAMZA-21:
----------------------------------
rb: https://reviews.apache.org/r/16706/
> Change KafkaSystemConsumer and BrokerProxy consumer defaults
> ------------------------------------------------------------
>
> Key: SAMZA-21
> URL: https://issues.apache.org/jira/browse/SAMZA-21
> Project: Samza
> Issue Type: Bug
> Components: kafka
> Affects Versions: 0.6.0
> Reporter: Chris Riccomini
> Assignee: Jakob Homan
> Labels: newbie
> Attachments: SAMZA-21.patch
>
>
> The Kafka BrokerProxy and KafkaSystemConsumer currently both have a socket
> timeout with a default of Int.MaxValue:
> {code}
> val timeout: Int = Int.MaxValue,
> {code}
> We don't actually use this default when we use the KafkaSystemFactory,
> because we use ConsumerConfig's default, which is 30 seconds. Nevertheless,
> we should change both classes to default to ConsumerConfig.SocketTimeout.
> While we're at it, we should also change bufferSize to be
> ConsumerConfig.SocketBufferSize, and we should change DefaultFetch to use:
> {code}
> val maxWait:Int = ConsumerConfig.MaxFetchWaitMs
> val minBytes:Int = ConsumerConfig.MinFetchBytes
> {code}
> We should also actually use maxWait in DefaultFetch.defaultFetch. Right now,
> we're hard coding 1000ms.
> Finally, BrokerProxy has the fetchSize hard coded:
> {code}
> val fetchSize: Int = 256 * 1024
> {code}
> We should make this parameter configurable
> (systems.%s.consumer.fetch.message.max.bytes), and pass it from the
> KafkaSystemFactory to the KafkaSystem to the BrokerProxy, just like timeout
> and buffer size.
> (ConsumerConfig is Kafka's config class)
--
This message was sent by Atlassian JIRA
(v6.1.5#6160)
