ads-bz 备份集群升级 3.9.2 操作手册

ads 1.1.1
  |
  +--> MM2 旧链路 --> ads-bz (1.1.1)（9110/9111，继续保留）
  |
  +--> MM2 新链路 --> ads-bz (3.9.2)（8110/8111，KRaft）

ads 1.1.1
  |
  +--> MM2 --> ads-bz (3.9.2)（9110/9111，KRaft）

mkdir -p /data0/logs/kafka-adbz

#!/bin/bash
# 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.

if [ $# -lt 1 ];
then
        echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
        exit 1
fi
base_dir=$(dirname $0)

if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx6G -Xms6G"          # Controller 使用 6G，broker 使用 16G
fi

EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}

COMMAND=$1
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    shift
    ;;
  *)
    ;;
esac

IP=`ifconfig | grep 'netmask 255.255.255' | grep inet | awk '{print $2}' | head -n1`
export JMX_PORT="9893"                               # 与 broker 保持一致
export PROMETHEUS_PORT="10013"                       # 与 broker 保持一致
export HADOOP_USER_NAME="hadoop_kraft"
export JAVA_HOME="/usr/local/jdk-17.0.7"
export LOG_DIR=/data0/logs/kafka-adbz                # Controller 与 broker 统一使用 kafka-adbz 日志目录
export KAFKA_OPTS="-Djava.security.auth.login.config=$base_dir/../config/kafka_server_jaas.conf"
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=true -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.host=0.0.0.0 -Djava.rmi.server.hostname=$IP -Dcom.sun.management.jmxremote.password.file=$base_dir/../config/jmxremote.password -Dcom.sun.management.jmxremote.access.file=$base_dir/../config/jmxremote.access -javaagent:$base_dir/../libs/jmx_prometheus_javaagent-0.20.0.jar=${PROMETHEUS_PORT:-19999}:$base_dir/../config/jmx.yaml"

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

mkdir -p /data1/kafka-controller-adbz/data

# 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.

#
# This configuration file is intended for use in KRaft mode, where
# Apache ZooKeeper is not present.
#

############################# Server Basics #############################

# The role of this server. Setting this puts us in KRaft mode
process.roles=controller

# The node id associated with this instance's roles
node.id=1000

# List of controller endpoints used connect to the controller cluster
controller.quorum.bootstrap.servers=kraft-adbz-controller01.kafka.data.sina.com.cn:8993,kraft-adbz-controller02.kafka.data.sina.com.cn:8993,kraft-adbz-controller03.kafka.data.sina.com.cn:8993,kraft-adbz-controller04.kafka.data.sina.com.cn:8993,kraft-adbz-controller05.kafka.data.sina.com.cn:8993

# inter.broker.protocol.version=3.9

# The max time that the client waits to establish a connection to ZooKeeper for authorizer
# authorizer.zookeeper.connection.timeout.ms=15000

# Zookeeper session timeout 
# authorizer.zookeeper.session.timeout.ms=60000

# Name of listener used for communication between brokers
# inter.broker.listener.name=SASL_PLAINTEXT

# SASL mechanism used for inter-broker communication.
# sasl.mechanism.inter.broker.protocol=PLAIN

# The fully qualified name of a class which is used by the broker for authorization.
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=false
kafka.authorizer.wait.timeout.ms=20000

############################# Socket Server Settings #############################

# The address the socket server listens on.
listeners=CONTROLLER://:8993

# Listener name, hostname and port the broker or the controller will advertise to clients.
# If not set, it uses the value for "listeners".
advertised.listeners=CONTROLLER://<本机IP>:8993

sasl.enabled.mechanisms=PLAIN

# A comma-separated list of the names of the listeners used by the controller.
# This is required if running in KRaft mode.
controller.listener.names=CONTROLLER

sasl.mechanism.controller.protocol=PLAIN

client.api.version.cap.enable=true

# Maps listener names to security protocols, the default is for them to be the same.
listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# super user 
super.users=User:admin

# The number of threads that the server uses for receiving requests from the network and sending responses to the network
num.network.threads=20

# The number of threads that the server uses for processing requests, which may include disk I/O
num.io.threads=20

# 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

# The number of queued requests allowed before blocking the network threads
queued.max.requests=200

############################# Log Basics #############################

# A comma separated list of directories under which to store log files
log.dirs=/data1/kafka-controller-adbz/data

# The default number of log partitions per topic.
num.partitions=3

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
# The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
offsets.topic.replication.factor=3

# The replication factor for the transaction topic 
transaction.state.log.replication.factor=3

# Overridden min.insync.replicas config for the transaction topic
transaction.state.log.min.isr=1

# Enable auto creation of topic on the server.
auto.create.topics.enable=false

# Enables auto leader balancing.
auto.leader.rebalance.enable=false

############################# 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.

# A size-based retention policy for logs.
log.retention.bytes=536870912000

# The maximum size of a log segment file.
log.segment.bytes=1073741824

# The minimum age of a log file to be eligible for deletion
log.retention.hours=8

# The maximum time before a new log segment is rolled out (in hours)
log.roll.hours=24

# Indicates whether to enable replicas not in the ISR set to be elected as leader as a last resort
unclean.leader.election.enable=true

# 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=300000

# ----- Enable tiered storage -----
remote.log.storage.system.enable=false

<node-id>@<hostname>:<port>:<directory-id>

# 只在发布机或任意一台 Controller 上执行一次。
# Controller IP 使用实际 IP，不使用域名。
CONTROLLER_1000_IP=<controller01-ip>
CONTROLLER_1001_IP=<controller02-ip>
CONTROLLER_1002_IP=<controller03-ip>
CONTROLLER_1003_IP=<controller04-ip>
CONTROLLER_1004_IP=<controller05-ip>

KAFKA_CLUSTER_ID=$(bin/kafka-storage.sh random-uuid)
CONTROLLER_1000_DIRECTORY_ID=$(bin/kafka-storage.sh random-uuid)
CONTROLLER_1001_DIRECTORY_ID=$(bin/kafka-storage.sh random-uuid)
CONTROLLER_1002_DIRECTORY_ID=$(bin/kafka-storage.sh random-uuid)
CONTROLLER_1003_DIRECTORY_ID=$(bin/kafka-storage.sh random-uuid)
CONTROLLER_1004_DIRECTORY_ID=$(bin/kafka-storage.sh random-uuid)

INITIAL_CONTROLLERS="1000@${CONTROLLER_1000_IP}:8993:${CONTROLLER_1000_DIRECTORY_ID},1001@${CONTROLLER_1001_IP}:8993:${CONTROLLER_1001_DIRECTORY_ID},1002@${CONTROLLER_1002_IP}:8993:${CONTROLLER_1002_DIRECTORY_ID},1003@${CONTROLLER_1003_IP}:8993:${CONTROLLER_1003_DIRECTORY_ID},1004@${CONTROLLER_1004_IP}:8993:${CONTROLLER_1004_DIRECTORY_ID}"

printf '%s\n' "$KAFKA_CLUSTER_ID" | tee ads-bz.cluster-id
printf '%s\n' "$INITIAL_CONTROLLERS" | tee ads-bz.initial-controllers

KAFKA_CLUSTER_ID=$(cat ads-bz.cluster-id)
INITIAL_CONTROLLERS=$(cat ads-bz.initial-controllers)

bin/kafka-storage.sh format \
  --initial-controllers "$INITIAL_CONTROLLERS" \
  -t "$KAFKA_CLUSTER_ID" \
  -c config/kraft/server.properties

bin/kafka-server-start.sh -daemon config/kraft/server.properties

mkdir -p /data0/logs/kafka-adbz

#!/bin/bash
# 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.

if [ $# -lt 1 ];
then
        echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
        exit 1
fi
base_dir=$(dirname $0)

if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx16G -Xms16G"
fi

EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}

COMMAND=$1
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    shift
    ;;
  *)
    ;;
esac

IP=`ifconfig | grep 'netmask 255.255.255' | grep inet | awk '{print $2}' | head -n1`
export JMX_PORT="9893"
export PROMETHEUS_PORT="10013"
export HADOOP_USER_NAME="hadoop_kraft"
export JAVA_HOME="/usr/local/jdk-17.0.7"
export LOG_DIR=/data0/logs/kafka-adbz
export KAFKA_OPTS="-Djava.security.auth.login.config=$base_dir/../config/kafka_server_jaas.conf"
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=true -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.host=0.0.0.0 -Djava.rmi.server.hostname=$IP -Dcom.sun.management.jmxremote.password.file=$base_dir/../config/jmxremote.password -Dcom.sun.management.jmxremote.access.file=$base_dir/../config/jmxremote.access -javaagent:$base_dir/../libs/jmx_prometheus_javaagent-0.20.0.jar=${PROMETHEUS_PORT:-19999}:$base_dir/../config/jmx.yaml"

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

mkdir -p /data1/kafka-adbz/data /data2/kafka-adbz/data /data3/kafka-adbz/data /data4/kafka-adbz/data /data5/kafka-adbz/data /data6/kafka-adbz/data /data7/kafka-adbz/data /data8/kafka-adbz/data /data9/kafka-adbz/data /data10/kafka-adbz/data

# 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.
node.id=1
process.roles=broker

############################# Socket Server Settings #############################
# List of comma-separated URIs the REST API will listen on.
listeners=SASL_PLAINTEXT_BROKER://:8111,SASL_PLAINTEXT://:8110
advertised.listeners=SASL_PLAINTEXT_BROKER://<本机IP>:8111,SASL_PLAINTEXT://<客户端-advertised-域名>:8110
# Map between listener names and security protocols.
listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,SASL_PLAINTEXT_BROKER:SASL_PLAINTEXT
# Name of listener used for communication between brokers
inter.broker.listener.name=SASL_PLAINTEXT_BROKER
# SASL mechanism used for inter-broker communication.
sasl.mechanism.inter.broker.protocol=PLAIN
# The list of SASL mechanisms enabled in the Kafka server.
sasl.enabled.mechanisms=PLAIN
# The super user does not require ACL authentication and has the highest permission
super.users=User:admin

############################# Broker Basic ###########################
# Rack of the broker.
broker.rack=rack-A
# The number of threads handling network requests
num.network.threads=40
# The number of threads doing disk I/O
num.io.threads=40
# 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
## The number of queued requests allowed before blocking the network threads
queued.max.requests=200
# The purge interval (in number of requests) of the fetch request purgatory
fetch.purgatory.purge.interval.requests=100
# The purge interval (in number of requests) of the producer request purgatory	
producer.purgatory.purge.interval.requests=100
# The amount of time the group coordinator will wait for more consumers to join a new group before performing the first rebalance
group.initial.rebalance.delay.ms=3000
# When set to true, the broker caps the maximum advertised and accepted version of the APIs listed in client.api.version.cap.overrides for client-facing listeners only.
client.api.version.cap.enable=true
# use the default configuration
# client.api.version.cap.overrides=METADATA:6,FETCH:8,LIST_OFFSETS:3,OFFSET_COMMIT:5,OFFSET_FETCH:4,OFFSET_FOR_LEADER_EPOCH:1

############################# acl for broker -- zookeeper #############################
# zookeeper.set.acl=true
allow.everyone.if.no.acl.found=false
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
# authorizer.zookeeper.connection.timeout.ms=15000
# authorizer.zookeeper.session.timeout.ms=60000
kafka.authorizer.wait.timeout.ms=20000

############################# Topic Basic ###########################
auto.create.topics.enable=false
auto.leader.rebalance.enable=false
## The frequency with which the partition rebalance check is triggered by the controller. set 600s
leader.imbalance.check.interval.seconds=3600
## The ratio of leader imbalance allowed per broker. The controller would trigger a leader balance if it goes above this value per broker. The value is specified in percentage.
leader.imbalance.per.broker.percentage=20
# Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off
delete.topic.enable=true
## Specify the final compression type for a given topic. This configuration accepts the standard compression codecs ('gzip', 'snappy', lz4).
## It additionally accepts 'uncompressed' which is equivalent to no compression; and 'producer' which means retain the original compression codec set by the producer.
compression.type=producer
## Any consumer distinguished by clientId/consumer group will get throttled if it fetches more bytes than this value per-second
#quota.consumer.default=20971520
default.replication.factor=3
min.insync.replicas=2

############################# Replication configurations ################################
# Number of fetcher threads used to replicate messages from a source broker.
num.replica.fetchers=8
# The number of bytes of messages to attempt to fetch for each partition. 
replica.fetch.max.bytes=1048576
# Minimum bytes expected for each fetch response. If not enough bytes, wait up to replicaMaxWaitTimeMs
replica.fetch.min.bytes=102400
# max wait time for each fetcher request issued by follower replicas.
replica.fetch.wait.max.ms=300
# The frequency with which the high watermark is saved out to disk
replica.high.watermark.checkpoint.interval.ms=3000
## If a follower hasn't sent any fetch requests or hasn't consumed up to the leaders log end offset for at least this time, the leader will remove the follower from isr
replica.lag.time.max.ms=10000
# The socket receive buffer for network requests
replica.socket.receive.buffer.bytes=65536
# The socket timeout for network requests. Its value should be at least replica.fetch.wait.max.ms	
replica.socket.timeout.ms=30000

############################# offset Basics ################################
# 	The number of partitions for the offset commit topic
offsets.topic.num.partitions=30
# The replication factor for the offsets topic
offsets.topic.replication.factor=3
# Offsets older than this retention period will be discarded
offsets.retention.minutes=2880
## The offsets topic segment bytes should be kept relatively small in order to facilitate faster log compaction and cache loads
offsets.topic.segment.bytes=67108864

############################# Log Basics #############################
# A comma seperated list of directories under which to store log files
log.dirs=/data1/kafka-adbz/data,/data2/kafka-adbz/data,/data3/kafka-adbz/data,/data4/kafka-adbz/data,/data5/kafka-adbz/data,/data6/kafka-adbz/data,/data7/kafka-adbz/data,/data8/kafka-adbz/data,/data9/kafka-adbz/data,/data10/kafka-adbz/data
# A comma seperated list of directories under which to store log files
log.cleanup.policy = delete
# Enable the log cleaner process to run on the server.
log.cleaner.enable=true
# 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=3
# The largest record batch size allowed by Kafka.
message.max.bytes=1000012
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# 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=100000
#log.flush.scheduler.interval.ms=5000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=5000
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760

############################# 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=72
# 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=1073741824
log.retention.bytes=536870912000
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1610612736
# The maximum time before a new log segment is rolled out (in hours), secondary to log.roll.ms property
log.roll.hours=24
# 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=300000
# The number of background threads to use for log cleaning
log.cleaner.threads=2

############################# 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: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=zk1.data.sina.com.cn:2181,zk6.data.sina.com.cn:2181,zk2.data.sina.com.cn:2181,zk5.data.sina.com.cn:2181,zk4.data.sina.com.cn:2181,zk3.data.sina.com.cn:2181,zk7.data.sina.com.cn:2181/weibo_ads_cap_test
# Timeout in ms for connecting to zookeeper
# zookeeper.connection.timeout.ms=10000
##session timeout also depend on the zookeeper config of minSessionTimeout and maxSessionTimeout
# zookeeper.session.timeout.ms=30000

############################# controller #############################
# Name of listener used for communication between controller and brokers
# control.plane.listener.name=CONTROLLER
# The socket timeout for controller-to-broker channels
controller.socket.timeout.ms=30000
# Enable controlled shutdown of the server	
controlled.shutdown.enable=true
controller.quorum.bootstrap.servers=kraft-adbz-controller01.kafka.data.sina.com.cn:8993,kraft-adbz-controller02.kafka.data.sina.com.cn:8993,kraft-adbz-controller03.kafka.data.sina.com.cn:8993,kraft-adbz-controller04.kafka.data.sina.com.cn:8993,kraft-adbz-controller05.kafka.data.sina.com.cn:8993
controller.listener.names=CONTROLLER
sasl.mechanism.controller.protocol=PLAIN
# zookeeper.metadata.migration.enable=true

############################# client #############################
client.fetchOrProduce.correlation.interval=1000
client.fetchOrProduce.correlation.init=20
client.heartbeat.correlation.interval=1000
client.heartbeat.correlation.init=20

############################# transaction #############################
# transaction.state.log.min.isr=1
# transaction.state.log.replication.factor=3
unclean.leader.election.enable=true
# 远程存储配置
############################# Log Tiered Storage ############################
# ----- Enable tiered storage -----
remote.log.storage.system.enable=false
remote.log.reader.threads=100
remote.log.reader.max.pending.tasks=10000
# Retention of remote log metadata topic in milliseconds.
# Notes:To avoid any data loss, this value should be more than the maximum retention period of any topic enabled with tiered storage in the cluster.
remote.log.metadata.topic.retention.ms=2592000000
# The maximum number of bytes that can be copied from local storage to remote storage per second
remote.log.manager.copy.max.bytes.per.second=209715200

# ----- Configure the remote log manager -----
# This is the default, but adding it for explicitness:
remote.log.metadata.manager.class.name=org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager

# Put the real listener name you'd like to use here:
remote.log.metadata.manager.listener.name=SASL_PLAINTEXT
rlmm.config.remote.log.metadata.common.client.sasl.mechanism=PLAIN
rlmm.config.remote.log.metadata.common.client.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='d4a12dfe3f97e641edd9f206eca5ae92';
rlmm.config.remote.log.metadata.producer.security.protocol=SASL_PLAINTEXT
rlmm.config.remote.log.metadata.producer.sasl.mechanism=PLAIN
rlmm.config.remote.log.metadata.producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='d4a12dfe3f97e641edd9f206eca5ae92';
rlmm.config.remote.log.metadata.consumer.security.protocol=SASL_PLAINTEXT
rlmm.config.remote.log.metadata.consumer.sasl.mechanism=PLAIN
rlmm.config.remote.log.metadata.consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='d4a12dfe3f97e641edd9f206eca5ae92';

# You may need to set this if you're experimenting with a single broker setup:
#rlmm.config.remote.log.metadata.topic.replication.factor=1

# ----- Configure the remote storage manager -----
# Here you need either one or two directories depending on what you did in Step 1:
remote.log.storage.manager.class.path=/usr/local/kafka/kafka-release/tired-storage/*
remote.log.storage.manager.class.name=io.aiven.kafka.tieredstorage.RemoteStorageManager
# 8 MiB is the current recommended chunk size:
rsm.config.chunk.size=8388608

# ----- Configure the storage backend -----
# Using HDFS backend as an example:
rsm.config.storage.backend.class=io.arenadata.kafka.tieredstorage.storage.hdfs.HdfsStorage
rsm.config.storage.hdfs.root=/user/hadoop_kraft/weibo_ads_cap_test
rsm.config.storage.hdfs.core-site.path=/usr/local/kafka/kafka-release/config/hadoop/core-site.xml
rsm.config.storage.hdfs.hdfs-site.path=/usr/local/kafka/kafka-release/config/hadoop/hdfs-site.xml
# The prefix can be skipped:
#rsm.config.storage.key.prefix: "some/prefix/"

# ----- Configure the fetch chunk cache -----
rsm.config.fetch.chunk.cache.class=io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCache
rsm.config.fetch.chunk.cache.path=/data11/cache
# fetch chunk rate
rsm.config.fetch.chunk.cache.write.rate.limit.bytes.per.second=524288000
# Pick some cache size, 300 GiB here:
rsm.config.fetch.chunk.cache.size=322122547200
# Prefetching size, 16 MiB here:
rsm.config.fetch.chunk.cache.prefetch.max.size=16777216
# Cache retention time ms, where -1 represents infinite retention
rsm.config.fetch.chunk.cache.retention.ms=600000

KAFKA_CLUSTER_ID=$(cat ads-bz.cluster-id)
bin/kafka-storage.sh format \
  -t "$KAFKA_CLUSTER_ID" \
  -c config/kraft/server.properties

bin/kafka-server-start.sh -daemon config/kraft/server.properties

bin/kafka-topics.sh \
  --bootstrap-server "$BOOTSTRAP" \
  --command-config "$ADMIN" \
  --create \
  --topic ads-bz-upgrade-canary \
  --partitions 3 \
  --replication-factor 3