anchor.diff:  # fail over, replica lag etc). This config determines the amount of time to wait before retrying.
   <%= render 'partials/option.erb', variables: {key: 'controlled.shutdown.retry.backoff.ms', value: node[:kafka][:controlled_shutdown][:retry_backoff_ms]} -%>
   
  -# enable controlled shutdown of the server
  +# Enable controlled shutdown of the server
   <%= render 'partials/option.erb', variables: {key: 'controlled.shutdown.enable', value: node[:kafka][:controlled_shutdown][:enabled]} -%>
   
   <% unless node[:kafka][:version] == '0.8.0' %>
  -############################# Leader imbalance configuration #############################
  +############################# Leader Imbalance Configuration #############################
   
   # If this is enabled the controller will automatically try to balance leadership for partitions
   # among the brokers by periodically returning leadership to the "preferred" replica for each
  
positive.code: <% end %>
  ############################# Replication Configuration #############################
  
  # the socket timeout for controller-to-broker channels
  <%= render 'partials/option.erb', variables: {key: 'controller.socket.timeout.ms', value: node[:kafka][:controller][:socket_timeout_ms]} -%>
  
  # the buffer size for controller-to-broker-channels
  <%= render 'partials/option.erb', variables: {key: 'controller.message.queue.size', value: node[:kafka][:controller][:message_queue_size]} -%>
  
  # default replication factor for automatically created topics
  <%= render 'partials/option.erb', variables: {key: 'default.replication.factor', value: node[:kafka][:default_replication_factor]} -%>
  
  # if a follower hasn't sent any fetch requests during this time, the leader will remove the follower from isr
  <%= render 'partials/option.erb', variables: {key: 'replica.lag.time.max.ms', value: node[:kafka][:replica][:lag_time_max_ms]} -%>
  
  # If the lag in messages between a leader and a follower exceeds this number, the leader will remove the follower from isr
  <%= render 'partials/option.erb', variables: {key: 'replica.lag.max.messages', value: node[:kafka][:replica][:lag_max_messages]} -%>
  
  # the socket timeout for network requests
  <%= render 'partials/option.erb', variables: {key: 'replica.socket.timeout.ms', value: node[:kafka][:replica][:socket_timeout_ms]} -%>
  
  # the socket receive buffer for network requests
  <%= render 'partials/option.erb', variables: {key: 'replica.socket.receive.buffer.bytes', value: node[:kafka][:replica][:socket_receive_buffer_bytes]} -%>
  
  # the number of bytes of messages to attempt to fetch
  <%= render 'partials/option.erb', variables: {key: 'replica.fetch.max.bytes', value: node[:kafka][:replica][:fetch_max_bytes]} -%>
  
  # minimum bytes expected for each fetch response. If not enough bytes, wait up to replicaMaxWaitTimeMs
  <%= render 'partials/option.erb', variables: {key: 'replica.fetch.min.bytes', value: node[:kafka][:replica][:fetch_min_bytes]} -%>
  
  # max wait time for each fetcher request issued by follower replicas
  <%= render 'partials/option.erb', variables: {key: 'replica.fetch.wait.max.ms', value: node[:kafka][:replica][:fetch_wait_max_ms]} -%>
  
  # number of fetcher threads used to replicate messages from a source broker.
  # Increasing this value can increase the degree of I/O parallelism in the follower broker.
  <%= render 'partials/option.erb', variables: {key: 'num.replica.fetchers', value: node[:kafka][:num_replica_fetchers]} -%>
  
  # the frequency with which the high watermark is saved out to disk
  <%= render 'partials/option.erb', variables: {key: 'replica.high.watermark.checkpoint.interval.ms', value: node[:kafka][:replica][:high_watermark_checkpoint_interval_ms]} -%>
  
  # the purge interval (in number of requests) of the fetch request purgatory
  <%= render 'partials/option.erb', variables: {key: 'fetch.purgatory.purge.interval.requests', value: node[:kafka][:fetch][:purgatory_purge_interval_requests]} -%>
  
  # the purge interval (in number of requests) of the producer request purgatory
  <%= render 'partials/option.erb', variables: {key: 'producer.purgatory.purge.interval.requests', value: node[:kafka][:producer][:purgatory_purge_interval_requests]} -%>
  
  ############################# Controlled shutdown configuration #############################
  
  # Controlled shutdown can fail for multiple reasons. This determines the number of retries when such failure happens
  <%= render 'partials/option.erb', variables: {key: 'controlled.shutdown.max.retries', value: node[:kafka][:controlled_shutdown][:max_retries]} -%>
  
negative.code: # The frequency with which to check for leader imbalance.
  <%= render 'partials/option.erb', variables: {key: 'leader.imbalance.check.interval.seconds', value: node[:kafka][:leader][:imbalance_check_interval_seconds]} -%>
  
  ############################# Consumer Offset Management #############################
  
  # The maximum amount of metadata to allow clients to save with their offsets.
  <%= render 'partials/option.erb', variables: {key: 'offset.metadata.max.bytes', value: node[:kafka][:offset_metadata_max_bytes]} -%>
  
  <% end %>
  ############################# ZooKeeper #############################
  
  # ZooKeeper connection string (see ZooKeeper docs for details).
  # This is a comma separated host:port pairs, each corresponding to a zk
  # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
  # You can also append an optional chroot string to the urls to specify the
  # root directory for all kafka znodes.
  <%= render 'partials/option.erb', variables: {key: 'zookeeper.connect', value: @zookeeper_connect} -%>
  
  # The max time that the client waits to establish a connection to ZooKeeper
  <%= render 'partials/option.erb', variables: {key: 'zookeeper.connection.timeout.ms', value: node[:kafka][:zookeeper][:connection_timeout_ms]} -%>
  
  # ZooKeeper session timeout
  <%= render 'partials/option.erb', variables: {key: 'zookeeper.session.timeout.ms', value: node[:kafka][:zookeeper][:session_timeout_ms]} -%>
  
  # how far a ZK follower can be behind a ZK leader
  