Merge pull request #1606 from dnwe/append-to-server-properties

fix: use modern kafka server.properties in vagrant

vagrant/boot_cluster.sh

@@ -17,6 +17,25 @@ while ! nc -q 1 localhost 21805 </dev/null; do echo "Waiting"; sleep 1; done
 # Launch and wait for Kafka
 for i in 1 2 3 4 5; do
     KAFKA_PORT=`expr $i + 9090`
-    cd ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT} && bin/kafka-server-start.sh -daemon config/server.properties
+    cd ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT} && bash bin/kafka-server-start.sh -daemon config/server.properties
 done
-while ! nc -q 1 localhost 29095 </dev/null; do echo "Waiting"; sleep 1; done
+ps auxww | grep -i kafka
+
+N=10
+RC=1
+set +x
+printf "Waiting for Kafka to become available."
+for _ in $(seq 1 "$N"); do
+  if nc -z 127.0.0.1 29095 </dev/null; then
+      RC=0
+      break
+  fi
+  printf "."
+  sleep 1
+done
+printf "\n"
+if [ "$RC" -gt 0 ]; then
+  echo 'Error: Kafka failed to startup' >&2
+  find "${KAFKA_INSTALL_ROOT}" -name "server.log" -print0 | xargs tail -256
+  exit ${RC}
+fi

vagrant/install_cluster.sh

@@ -16,25 +16,62 @@ rm -f ${KAFKA_INSTALL_ROOT}/toxiproxy
 ln -s ${KAFKA_INSTALL_ROOT}/toxiproxy-${TOXIPROXY_VERSION} ${KAFKA_INSTALL_ROOT}/toxiproxy
 
 for i in 1 2 3 4 5; do
-    ZK_PORT=`expr $i + 2180`
-    ZK_PORT_REAL=`expr $i + 21800`
-    KAFKA_PORT=`expr $i + 9090`
-    KAFKA_PORT_REAL=`expr $i + 29090`
+    ZK_PORT=$((i + 2180))
+    ZK_PORT_REAL=$((i + 21800))
+    KAFKA_PORT=$((i + 9090))
+    KAFKA_PORT_REAL=$((i + 29090))
 
     # unpack kafka
     mkdir -p ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}
     tar xzf ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_VERSION}.tgz -C ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT} --strip-components 1
 
     # broker configuration
-    cp ${REPOSITORY_ROOT}/vagrant/server.properties ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/
-    sed -i s/KAFKAID/${KAFKA_PORT}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-    sed -i s/KAFKAPORT/${KAFKA_PORT_REAL}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-    sed -i s/KAFKA_HOSTNAME/${KAFKA_HOSTNAME}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-    sed -i s/ZK_PORT/${ZK_PORT}/g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
-
-    KAFKA_DATADIR="${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/data"
-    mkdir -p ${KAFKA_DATADIR}
-    sed -i s#KAFKA_DATADIR#${KAFKA_DATADIR}#g ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties
+    mkdir -p "${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/data"
+
+    # Append to default server.properties with a small number of customisations
+    printf "\n\n" >> "${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties"
+    cat << EOF >> "${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/server.properties"
+############################# Sarama Test Cluster #############################
+
+broker.id=${KAFKA_PORT}
+broker.rack=${i}
+
+# Listen on "real" port
+listeners=PLAINTEXT://:${KAFKA_PORT_REAL}
+# Advertise Toxiproxy port
+advertised.listeners=PLAINTEXT://${KAFKA_HOSTNAME}:${KAFKA_PORT}
+
+# Connect to Zookeeper via Toxiproxy port
+zookeeper.connect=127.0.0.1:${ZK_PORT}
+
+# Data directory
+log.dirs="${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/data"
+
+# Create new topics with a replication factor of 2 so failover can be tested
+# more easily.
+default.replication.factor=2
+
+# Turn on log.retention.bytes to avoid filling up the VM's disk
+log.retention.bytes=268435456
+log.segment.bytes=268435456
+
+# Enable topic deletion and disable auto-creation
+delete.topic.enable=true
+auto.create.topics.enable=false
+
+# Lower the zookeeper timeouts so we don't have to wait forever for a node
+# to die when we use toxiproxy to kill its zookeeper connection
+zookeeper.session.timeout.ms=3000
+zookeeper.connection.timeout.ms=3000
+
+# Disable broker ID length constraint
+reserved.broker.max.id=10000
+
+# Permit follower fetching (KIP-392)
+replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector
+
+###############################################################################
+EOF
 
     # zookeeper configuration
     cp ${REPOSITORY_ROOT}/vagrant/zookeeper.properties ${KAFKA_INSTALL_ROOT}/kafka-${KAFKA_PORT}/config/

vagrant/server.properties

@@ -1,127 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-# see kafka.server.KafkaConfig for additional details and defaults
-
-############################# Server Basics #############################
-
-# The id of the broker. This must be set to a unique integer for each broker.
-broker.id=KAFKAID
-reserved.broker.max.id=10000
-
-############################# Socket Server Settings #############################
-
-# The port the socket server listens on
-port=KAFKAPORT
-
-# Hostname the broker will bind to. If not set, the server will bind to all interfaces
-host.name=localhost
-
-# Hostname the broker will advertise to producers and consumers. If not set, it uses the
-# value for "host.name" if configured.  Otherwise, it will use the value returned from
-# java.net.InetAddress.getCanonicalHostName().
-advertised.host.name=KAFKA_HOSTNAME
-advertised.port=KAFKAID
-
-# The port to publish to ZooKeeper for clients to use. If this is not set,
-# it will publish the same port that the broker binds to.
-# advertised.port=<port accessible by clients>
-
-# The number of threads handling network requests
-num.network.threads=2
-
-# The number of threads doing disk I/O
-num.io.threads=8
-
-# The send buffer (SO_SNDBUF) used by the socket server
-socket.send.buffer.bytes=1048576
-
-# The receive buffer (SO_RCVBUF) used by the socket server
-socket.receive.buffer.bytes=1048576
-
-# The maximum size of a request that the socket server will accept (protection against OOM)
-socket.request.max.bytes=104857600
-
-
-############################# Log Basics #############################
-
-# A comma seperated list of directories under which to store log files
-log.dirs=KAFKA_DATADIR
-
-# The default number of log partitions per topic. More partitions allow greater
-# parallelism for consumption, but this will also result in more files across
-# the brokers.
-num.partitions=2
-
-# Create new topics with a replication factor of 2 so failover can be tested
-# more easily.
-default.replication.factor=2
-
-auto.create.topics.enable=false
-delete.topic.enable=true
-
-############################# Log Flush Policy #############################
-
-# Messages are immediately written to the filesystem but by default we only fsync() to sync
-# the OS cache lazily. The following configurations control the flush of data to disk.
-# There are a few important trade-offs here:
-#    1. Durability: Unflushed data may be lost if you are not using replication.
-#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
-#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
-# The settings below allow one to configure the flush policy to flush data after a period of time or
-# every N messages (or both). This can be done globally and overridden on a per-topic basis.
-
-# The number of messages to accept before forcing a flush of data to disk
-#log.flush.interval.messages=10000
-
-# The maximum amount of time a message can sit in a log before we force a flush
-#log.flush.interval.ms=1000
-
-############################# Log Retention Policy #############################
-
-# The following configurations control the disposal of log segments. The policy can
-# be set to delete segments after a period of time, or after a given size has accumulated.
-# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
-# from the end of the log.
-
-# The minimum age of a log file to be eligible for deletion
-log.retention.hours=168
-
-# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
-# segments don't drop below log.retention.bytes.
-log.retention.bytes=268435456
-
-# The maximum size of a log segment file. When this size is reached a new log segment will be created.
-log.segment.bytes=268435456
-
-# The interval at which log segments are checked to see if they can be deleted according
-# to the retention policies
-log.retention.check.interval.ms=60000
-
-# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
-# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
-log.cleaner.enable=false
-
-############################# 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.
-zookeeper.connect=localhost:ZK_PORT
-
-# Timeout in ms for connecting to zookeeper
-zookeeper.session.timeout.ms=3000
-zookeeper.connection.timeout.ms=3000