Using Streams for Apache Kafka on RHEL with ZooKeeper
Using Streams for Apache Kafka on RHEL with ZooKeeper Red Hat Streams for Apache Kafka 2.9 Configure and manage a deployment of Streams for Apache Kafka 2.9 on Red Hat Enterprise Linux
bin/kafka-console-consumer.sh --bootstrap-server <broker_host>:<port> --topic <topic_name> --from-beginning groupadd kafka useradd -g kafka kafka passwd kafka unzip amq-streams-<version>-kafka-bin.zip -d /opt mv /opt/kafka*redhat* /opt/kafka chown -R kafka:kafka /opt/kafka mkdir /var/lib/zookeeper chown -R kafka:kafka /var/lib/zookeeper mkdir /var/lib/kafka chown -R kafka:kafka /var/lib/kafka bootstrap.servers=localhost:9092 bootstrap.servers=kafka0.<host_ip_address>:9092,kafka1.<host_ip_address>:9092,kafka2.<host_ip_address>:9092 ./bin/kafka-storage.sh random-uuid log.dirs=/var/lib/kafka/ ./bin/kafka-server-start.sh -daemon ./config/server.properties jcmd | grep kafka process ID kafka.Kafka ./config/server.properties ./bin/kafka-console-producer.sh --broker-list <bootstrap_address> --topic <topic-name> ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic >message 1 >message 2 >message 3 >message 4 WARN Error while fetching metadata with correlation id 39 : {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) ./bin/kafka-console-consumer.sh --bootstrap-server <bootstrap_address> --topic <topic-name> --from-beginning ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning ./bin/kafka-server-stop.sh jcmd | grep kafka ./bin/zookeeper-server-stop.sh echo " <NodeID> " > /var/lib/zookeeper/myid echo "1" > /var/lib/zookeeper/myid tickTime=2000 dataDir=/var/lib/zookeeper/ initLimit=5 syncLimit=2 reconfigEnabled=true standaloneEnabled=false server.1=172.17.0.1:2888:3888:participant;172.17.0.1:2181 server.2=172.17.0.2:2888:3888:participant;172.17.0.2:2181 server.3=172.17.0.3:2888:3888:participant;172.17.0.3:2181 server.4=172.17.0.4:2888:3888:participant;172.17.0.4:2181 server.5=172.17.0.5:2888:3888:participant;172.17.0.5:2181 ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties jcmd | grep zookeeper number org.apache.zookeeper.server.quorum.QuorumPeerMain ./config/zookeeper.properties echo stat | ncat localhost 2181 ZooKeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 00:39 GMT Clients: /0:0:0:0:0:0:0:1:59726[0](queued=0,recved=1,sent=0) Latency min/avg/max: 0/0/0 Received: 2 Sent: 1 Connections: 1 Outstanding: 0 Zxid: 0x200000000 Mode: follower Node count: 4 broker.id=0 zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181 listeners=REPLICATION://:9091,PLAINTEXT://:9092 listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT inter.broker.listener.name=REPLICATION log.dirs=/var/lib/kafka ./bin/kafka-server-start.sh -daemon ./config/server.properties jcmd | grep Kafka process ID kafka.Kafka ./config/server.properties echo dump | ncat zoo1.my-domain.com 2181 SessionTracker dump: org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9 ephemeral nodes dump: Sessions with Ephemerals (3): 0x20000015dd00000: /brokers/ids/1 0x10000015dc70000: /controller /brokers/ids/0 0x10000015dc70001: /brokers/ids/2 ./bin/kafka-server-stop.sh ./bin/kafka-server-start.sh -daemon ./config/server.properties jcmd | grep kafka process ID kafka.Kafka ./config/server.properties echo dump | ncat zoo1.my-domain.com 2181 SessionTracker dump: org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9 ephemeral nodes dump: Sessions with Ephemerals (3): 0x20000015dd00000: /brokers/ids/1 0x10000015dc70000: /controller /brokers/ids/0 0x10000015dc70001: /brokers/ids/2 ./bin/kafka-topics.sh --bootstrap-server <bootstrap_address> --describe --under-replicated-partitions ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions Topic: topic3 Partition: 4 Leader: 2 Replicas: 2,3 Isr: 2 Topic: topic3 Partition: 5 Leader: 3 Replicas: 1,2 Isr: 1 Topic: topic1 Partition: 1 Leader: 3 Replicas: 1,3 Isr: 3 ... kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions [2022-06-08 14:32:29,885] INFO Terminating process due to signal SIGTERM (org.apache.kafka.common.utils.LoggingSignalHandler) [2022-06-08 14:32:29,886] INFO [KafkaServer id=0] shutting down (kafka.server.KafkaServer) [2022-06-08 14:32:29,887] INFO [KafkaServer id=0] Starting controlled shutdown (kafka.server.KafkaServer) [2022-06-08 14:32:29,896] INFO [KafkaServer id=0] Controlled shutdown request returned successfully after 6ms (kafka.server.KafkaServer) [2022-06-08 14:39:35,245] INFO [KafkaServer id=0] started (kafka.server.KafkaServer) <option> = <value> This is a comment sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="bob" password="bobs-password"; config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider option=USD{env: <MY_ENV_VAR_NAME> } tickTime=2000 dataDir=/var/lib/zookeeper/ clientPort=2181 tickTime=2000 dataDir=/var/lib/zookeeper/ initLimit=5 syncLimit=2 reconfigEnabled=true standaloneEnabled=false server.1=172.17.0.1:2888:3888:participant;172.17.0.1:2181 server.2=172.17.0.2:2888:3888:participant;172.17.0.2:2181 server.3=172.17.0.3:2888:3888:participant;172.17.0.3:2181 ContextName { param1 param2; }; QuorumServer { org.apache.zookeeper.server.auth.DigestLoginModule required user_zookeeper="123456"; }; QuorumLearner { org.apache.zookeeper.server.auth.DigestLoginModule required username="zookeeper" password="123456"; }; quorum.auth.enableSasl=true quorum.auth.learnerRequireSasl=true quorum.auth.serverRequireSasl=true quorum.auth.learner.loginContext=QuorumLearner quorum.auth.server.loginContext=QuorumServer quorum.cnxn.threads.size=20 export KAFKA_OPTS="-Djava.security.auth.login.config=./config/zookeeper-jaas.conf"; ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties Server { org.apache.zookeeper.server.auth.DigestLoginModule required user_super="123456" user_kafka="123456" user_someoneelse="123456"; }; requireClientAuthScheme=sasl authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider authProvider.2=org.apache.zookeeper.server.auth.SASLAuthenticationProvider authProvider.3=org.apache.zookeeper.server.auth.SASLAuthenticationProvider export KAFKA_OPTS="-Djava.security.auth.login.config=./config/zookeeper-jaas.conf"; ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties QuorumServer { org.apache.zookeeper.server.auth.DigestLoginModule required user_ <Username> =" <Password> "; }; QuorumLearner { org.apache.zookeeper.server.auth.DigestLoginModule required username=" <Username> " password=" <Password> "; }; QuorumServer { org.apache.zookeeper.server.auth.DigestLoginModule required user_zookeeper="123456"; }; QuorumLearner { org.apache.zookeeper.server.auth.DigestLoginModule required username="zookeeper" password="123456"; }; quorum.auth.enableSasl=true quorum.auth.learnerRequireSasl=true quorum.auth.serverRequireSasl=true quorum.auth.learner.loginContext=QuorumLearner quorum.auth.server.loginContext=QuorumServer quorum.cnxn.threads.size=20 export KAFKA_OPTS="-Djava.security.auth.login.config=./config/zookeeper-jaas.conf"; ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties Server { org.apache.zookeeper.server.auth.DigestLoginModule required user_super=" <SuperUserPassword> " user <Username1>_=" <Password1> " user <USername2>_=" <Password2> "; }; Server { org.apache.zookeeper.server.auth.DigestLoginModule required user_super="123456" user_kafka="123456"; }; requireClientAuthScheme=sasl authProvider. <IdOfBroker1> =org.apache.zookeeper.server.auth.SASLAuthenticationProvider authProvider. <IdOfBroker2> =org.apache.zookeeper.server.auth.SASLAuthenticationProvider authProvider. <IdOfBroker3> =org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthScheme=sasl authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider authProvider.2=org.apache.zookeeper.server.auth.SASLAuthenticationProvider authProvider.3=org.apache.zookeeper.server.auth.SASLAuthenticationProvider export KAFKA_OPTS="-Djava.security.auth.login.config=./config/zookeeper-jaas.conf"; ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181 zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181/my-cluster-1 <listener_name>://<hostname>:<port> listeners=internal-1://:9092,internal-2://:9093,replication://:9094 listeners=internal-1://:9092,internal-2://:9093 advertised.listeners=internal-1://my-broker-1.my-domain.com:1234,internal-2://my-broker-1.my-domain.com:1235 listeners=REPLICATION://0.0.0.0:9091 inter.broker.listener.name=REPLICATION listeners=CONTROLLER://0.0.0.0:9090 controller.listener.names=CONTROLLER controller.quorum.bootstrap.servers=localhost:9090 log.dirs=/var/lib/kafka log.dirs=/var/lib/kafka1,/var/lib/kafka2,/var/lib/kafka3 broker.id=1 listeners=INT1://:9092,INT2://:9093,REPLICATION://:9094 listener.security.protocol.map=INT1:SASL_PLAINTEXT,INT2:SASL_SSL,REPLICATION:SSL listener.security.protocol.map=INT1:SSL,INT2:SSL,REPLICATION:SSL ssl.keystore.location=/path/to/keystore/server-1.jks ssl.keystore.password=123456 listeners=INT1://:9092,INT2://:9093,REPLICATION://:9094 listener.security.protocol.map=INT1:SSL,INT2:SSL,REPLICATION:SSL Default configuration - will be used for listeners INT1 and INT2 ssl.keystore.location=/path/to/keystore/server-1.jks ssl.keystore.password=123456 Different configuration for listener REPLICATION listener.name.replication.ssl.keystore.location=/path/to/keystore/replication.jks listener.name.replication.ssl.keystore.password=123456 listeners=UNENCRYPTED://:9092,ENCRYPTED://:9093,REPLICATION://:9094 listener.security.protocol.map=UNENCRYPTED:PLAINTEXT,ENCRYPTED:SSL,REPLICATION:PLAINTEXT ssl.keystore.location=/path/to/keystore/server-1.jks ssl.keystore.password=123456 ssl.truststore.location=/path/to/truststore.jks ssl.truststore.password=123456 ssl.client.auth=required KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required user_admin="123456" user_user1="123456" user_user2="123456"; }; listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094 listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT sasl.enabled.mechanisms=PLAIN export KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/kafka-server-start.sh -daemon ./config/server.properties KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required; }; listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094 listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT sasl.enabled.mechanisms=SCRAM-SHA-512 export KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/kafka-server-start.sh -daemon ./config/server.properties KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required user_admin="123456" user_user1="123456" user_user2="123456"; com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/etc/security/keytabs/kafka_server.keytab" principal="kafka/kafka1.hostname.com@EXAMPLE.COM"; org.apache.kafka.common.security.scram.ScramLoginModule required; }; sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512 bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password=changeit]' --entity-type users --entity-name kafka sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 KafkaServer { org.apache.kafka.common.security.plain.ScramLoginModule required username="admin" password="123456" # }; ./bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --alter --add-config 'SCRAM-SHA-512=[password=<password>]' --entity-type users --entity-name <username> ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name user1 --bootstrap-server <broker_host>:<port> --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name <username> --bootstrap-server localhost:9092 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name user1 zookeeper.connect= node1.example.redhat.com :2181 /opt/kafka/krb5/zookeeper-node1.keytab /opt/kafka/krb5/kafka-node1.keytab /opt/kafka/krb5/kafka-producer1.keytab /opt/kafka/krb5/kafka-consumer1.keytab Client { com.sun.security.auth.module.Krb5LoginModule required debug=true useKeyTab=true 1 storeKey=true 2 useTicketCache=false 3 keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 4 principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 5 }; Server { com.sun.security.auth.module.Krb5LoginModule required debug=true useKeyTab=true storeKey=true useTicketCache=false keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; }; QuorumServer { com.sun.security.auth.module.Krb5LoginModule required debug=true useKeyTab=true storeKey=true keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; }; QuorumLearner { com.sun.security.auth.module.Krb5LoginModule required debug=true useKeyTab=true storeKey=true keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; }; requireClientAuthScheme=sasl jaasLoginRenew=3600000 1 kerberos.removeHostFromPrincipal=false 2 kerberos.removeRealmFromPrincipal=false 3 quorum.auth.enableSasl=true 4 quorum.auth.learnerRequireSasl=true 5 quorum.auth.serverRequireSasl=true quorum.auth.learner.loginContext=QuorumLearner 6 quorum.auth.server.loginContext=QuorumServer quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 7 quorum.cnxn.threads.size=20 export EXTRA_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties KafkaServer { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/opt/kafka/krb5/kafka-node1.keytab" principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; }; KafkaClient { com.sun.security.auth.module.Krb5LoginModule required debug=true useKeyTab=true storeKey=true useTicketCache=false keyTab="/opt/kafka/krb5/kafka-node1.keytab" principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; }; broker.id=0 listeners=SECURE://:9092,REPLICATION://:9094 1 inter.broker.listener.name=REPLICATION listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 2 .. sasl.enabled.mechanisms=GSSAPI 3 sasl.mechanism.inter.broker.protocol=GSSAPI 4 sasl.kerberos.service.name=kafka 5 export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties sasl.mechanism=GSSAPI 1 security.protocol=SASL_PLAINTEXT 2 sasl.kerberos.service.name=kafka 3 sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 4 useKeyTab=true useTicketCache=false storeKey=true keyTab="/opt/kafka/krb5/producer1.keytab" principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; sasl.mechanism=GSSAPI security.protocol=SASL_PLAINTEXT sasl.kerberos.service.name=kafka sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true useTicketCache=false storeKey=true keyTab="/opt/kafka/krb5/consumer1.keytab" principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties --topic topic1 --bootstrap-server node1.example.redhat.com:9094 export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties --topic topic1 --bootstrap-server node1.example.redhat.com:9094 authorizer.class.name=kafka.security.authorizer.AclAuthorizer super.users=User:admin,User:operator opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --operation Read --topic myTopic --allow-principal User:user1 --allow-principal User:user2 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --operation Describe --topic myTopic --allow-principal User:user1 --allow-principal User:user2 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --operation Read --operation Describe --group MyConsumerGroup --allow-principal User:user1 --allow-principal User:user2 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --operation Describe --operation Read --topic myTopic --group MyConsumerGroup --deny-principal User:user1 --deny-host 127.0.0.1 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --consumer --topic myTopic --group MyConsumerGroup --allow-principal User:user1 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --list --topic myTopic Current ACLs for resource `Topic:myTopic`: User:user1 has Allow permission for operations: Read from hosts: * User:user2 has Allow permission for operations: Read from hosts: * User:user2 has Deny permission for operations: Read from hosts: 127.0.0.1 User:user1 has Allow permission for operations: Describe from hosts: * User:user2 has Allow permission for operations: Describe from hosts: * User:user2 has Deny permission for operations: Describe from hosts: 127.0.0.1 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --remove --operation Read --topic myTopic --allow-principal User:user1 --allow-principal User:user2 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --remove --operation Describe --topic myTopic --allow-principal User:user1 --allow-principal User:user2 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --remove --operation Read --operation Describe --group MyConsumerGroup --allow-principal User:user1 --allow-principal User:user2 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --remove --consumer --topic myTopic --group MyConsumerGroup --allow-principal User:user1 opt/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --remove --operation Describe --operation Read --topic myTopic --group MyConsumerGroup --deny-principal User:user1 --deny-host 127.0.0.1 Client { org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="123456"; }; Client { org.apache.kafka.common.security.plain.PlainLoginModule required username=" <Username> " password=" <Password> "; }; Client { org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="123456"; }; export KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/kafka-server-start.sh -daemon ./config/server.properties zookeeper.set.acl=true zookeeper.set.acl=true zookeeper.set.acl=true KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=<zookeeper_url> sasl.enabled.mechanisms=OAUTHBEARER 1 listeners=CLIENT://0.0.0.0:9092 2 listener.security.protocol.map=CLIENT:SASL_PLAINTEXT 3 listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER 4 sasl.mechanism.inter.broker.protocol=OAUTHBEARER 5 inter.broker.listener.name=CLIENT 6 listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler 7 listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule # listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ 1 oauth.valid.issuer.uri="https://<auth_server_address>/<issuer-context>" \ 2 oauth.jwks.endpoint.uri="https://<oauth_server_address>/<path_to_jwks_endpoint>" \ 3 oauth.jwks.refresh.seconds="300" \ 4 oauth.jwks.refresh.min.pause.seconds="1" \ 5 oauth.jwks.expiry.seconds="360" \ 6 oauth.username.claim="preferred_username" \ 7 oauth.ssl.truststore.location="<path_to_truststore_p12_file>" \ 8 oauth.ssl.truststore.password="<truststore_password>" \ 9 oauth.ssl.truststore.type="PKCS12" ; 10 listener.name.client.oauthbearer.connections.max.reauth.ms=3600000 11 listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.introspection.endpoint.uri="https://<oauth_server_address>/<introspection_endpoint>" \ 1 oauth.client.id="kafka-broker" \ 2 oauth.client.secret="kafka-broker-secret" \ 3 oauth.ssl.truststore.location="<path_to_truststore_p12_file>" \ 4 oauth.ssl.truststore.password="<truststore_password>" \ 5 oauth.ssl.truststore.type="PKCS12" \ 6 oauth.username.claim="preferred_username" ; 7 listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required # oauth.token.endpoint.uri="https://<auth_server_address>/<path_to_token_endpoint>" \ 1 oauth.custom.claim.check="@.custom == 'custom-value'" \ 2 oauth.scope="<scope>" \ 3 oauth.check.audience="true" \ 4 oauth.audience="<audience>" \ 5 oauth.client.id="kafka-broker" \ 6 oauth.client.secret="kafka-broker-secret" \ 7 oauth.connect.timeout.seconds=60 \ 8 oauth.read.timeout.seconds=60 \ 9 oauth.http.retries=2 \ 10 oauth.http.retry.pause.millis=300 \ 11 oauth.groups.claim="USD.groups" \ 12 oauth.groups.claim.delimiter="," \ 13 oauth.include.accept.header="false" ; 14 oauth.check.issuer=false \ 15 oauth.username.prefix="user-account-" \ 16 oauth.fallback.username.claim="client_id" \ 17 oauth.fallback.username.prefix="service-account-" \ 18 oauth.valid.token.type="bearer" \ 19 oauth.userinfo.endpoint.uri="https://<auth_server_address>/<path_to_userinfo_endpoint>" ; 20 sasl.enabled.mechanisms=OAUTHBEARER listeners=CLIENT://0.0.0.0:9092 listener.security.protocol.map=CLIENT:SASL_PLAINTEXT listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER sasl.mechanism.inter.broker.protocol=OAUTHBEARER inter.broker.listener.name=CLIENT listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ 1 oauth.valid.issuer.uri="https://<auth_server_address>/<issuer-context>" oauth.jwks.endpoint.uri="https://<oauth_server_address>/<path_to_jwks_endpoint>" oauth.username.claim="preferred_username" oauth.client.id="kafka-broker" \ 2 oauth.client.secret="kafka-secret" \ 3 oauth.token.endpoint.uri="https://<oauth_server_address>/<token_endpoint>" ; 4 listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler 5 listener.name.client.oauthbearer.connections.max.reauth.ms=3600000 sasl.enabled.mechanisms=OAUTHBEARER listeners=REPLICATION://kafka:9091,CLIENT://kafka:9092 1 listener.security.protocol.map=REPLICATION:SSL,CLIENT:SASL_PLAINTEXT 2 listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER inter.broker.listener.name=REPLICATION listener.name.replication.ssl.keystore.password=<keystore_password> 3 listener.name.replication.ssl.truststore.password=<truststore_password> listener.name.replication.ssl.keystore.type=JKS listener.name.replication.ssl.truststore.type=JKS listener.name.replication.ssl.secure.random.implementation=SHA1PRNG 4 listener.name.replication.ssl.endpoint.identification.algorithm=HTTPS 5 listener.name.replication.ssl.keystore.location=<path_to_keystore> 6 listener.name.replication.ssl.truststore.location=<path_to_truststore> 7 listener.name.replication.ssl.client.auth=required 8 listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.valid.issuer.uri="https://<auth_server_address>/<issuer-context>" oauth.jwks.endpoint.uri="https://<oauth_server_address>/<path_to_jwks_endpoint>" oauth.username.claim="preferred_username" ; listeners=CLIENT://0.0.0.0:9092 listener.security.protocol.map=CLIENT:SASL_PLAINTEXT listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER,PLAIN sasl.mechanism.inter.broker.protocol=OAUTHBEARER inter.broker.listener.name=CLIENT listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.valid.issuer.uri="https:<auth_server_address>/<issuer-context>" oauth.jwks.endpoint.uri="https://<auth_server>/<path_to_jwks_endpoint>" oauth.username.claim="preferred_username" oauth.client.id="kafka-broker" oauth.client.secret="kafka-secret" oauth.token.endpoint.uri="https://<oauth_server_address>/<token_endpoint>" ; listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler 1 listener.name.client.plain.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.plain.JaasServerOauthOverPlainValidatorCallbackHandler 2 listener.name.client.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ 3 oauth.valid.issuer.uri="https://<auth_server_address>/<issuer-context>" oauth.jwks.endpoint.uri="https://<oauth_server_address>/<path_to_jwks_endpoint>" oauth.username.claim="preferred_username" oauth.token.endpoint.uri="https://<oauth_server_address>/<token_endpoint>" ; 4 listener.name.client.oauthbearer.connections.max.reauth.ms=3600000 security.protocol=SASL_SSL 1 sasl.mechanism=OAUTHBEARER 2 ssl.truststore.location=/tmp/truststore.p12 3 ssl.truststore.password=USDSTOREPASS ssl.truststore.type=PKCS12 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.token.endpoint.uri="<token_endpoint_url>" \ 4 oauth.client.id="<client_id>" \ 5 oauth.client.secret="<client_secret>" \ 6 oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 7 oauth.ssl.truststore.password="USDSTOREPASS" \ 8 oauth.ssl.truststore.type="PKCS12" \ 9 oauth.scope="<scope>" \ 10 oauth.audience="<audience>" ; 11 sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler security.protocol=SASL_SSL sasl.mechanism=OAUTHBEARER ssl.truststore.location=/tmp/truststore.p12 ssl.truststore.password=USDSTOREPASS ssl.truststore.type=PKCS12 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.token.endpoint.uri="<token_endpoint_url>" oauth.client.id="<client_id>" oauth.client.assertion.location="<path_to_client_assertion_token_file>" \ 1 oauth.client.assertion.type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer" \ 2 oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" oauth.ssl.truststore.password="USDSTOREPASS" oauth.ssl.truststore.type="PKCS12" oauth.scope="<scope>" oauth.audience="<audience>" ; sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler security.protocol=SASL_SSL sasl.mechanism=OAUTHBEARER ssl.truststore.location=/tmp/truststore.p12 ssl.truststore.password=USDSTOREPASS ssl.truststore.type=PKCS12 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.token.endpoint.uri="<token_endpoint_url>" oauth.client.id="<client_id>" \ 1 oauth.client.secret="<client_secret>" \ 2 oauth.password.grant.username="<username>" \ 3 oauth.password.grant.password="<password>" \ 4 oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" oauth.ssl.truststore.password="USDSTOREPASS" oauth.ssl.truststore.type="PKCS12" oauth.scope="<scope>" oauth.audience="<audience>" ; sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler security.protocol=SASL_SSL sasl.mechanism=OAUTHBEARER ssl.truststore.location=/tmp/truststore.p12 ssl.truststore.password=USDSTOREPASS ssl.truststore.type=PKCS12 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.access.token="<access_token>" ; 1 sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler security.protocol=SASL_SSL sasl.mechanism=OAUTHBEARER ssl.truststore.location=/tmp/truststore.p12 ssl.truststore.password=USDSTOREPASS ssl.truststore.type=PKCS12 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.access.token.location="/var/run/secrets/kubernetes.io/serviceaccount/token"; 1 sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler security.protocol=SASL_SSL sasl.mechanism=OAUTHBEARER ssl.truststore.location=/tmp/truststore.p12 ssl.truststore.password=USDSTOREPASS ssl.truststore.type=PKCS12 sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.token.endpoint.uri="<token_endpoint_url>" oauth.client.id="<client_id>" \ 1 oauth.client.secret="<client_secret>" \ 2 oauth.refresh.token="<refresh_token>" \ 3 oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" oauth.ssl.truststore.password="USDSTOREPASS" oauth.ssl.truststore.type="PKCS12" ; sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler oauth.sasl.extension.key1="value1" oauth.sasl.extension.key2="value2" sasl.enabled.mechanisms=OAUTHBEARER listeners=CLIENT://0.0.0.0:9092 listener.security.protocol.map=CLIENT:SASL_PLAINTEXT listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER sasl.mechanism.inter.broker.protocol=OAUTHBEARER inter.broker.listener.name=CLIENT listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ; listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required # oauth.client.id="kafka-broker" oauth.client.secret="kafka-broker-secret" oauth.ssl.truststore.location="<path_to_truststore_p12_file>" oauth.ssl.truststore.password="<truststore_password>" oauth.ssl.truststore.type="PKCS12" ; oauth.ssl.endpoint.identification.algorithm="" <dependency> <groupId>io.strimzi</groupId> <artifactId>kafka-oauth-client</artifactId> <version>0.15.0.redhat-00012</version> </dependency> Properties props = new Properties(); try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) { props.load(reader); } authorizer.class.name=io.strimzi.kafka.oauth.server.authorizer.KeycloakAuthorizer principal.builder.class=io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder strimzi.authorization.token.endpoint.uri="https://<auth_server_address>/auth/realms/REALM-NAME/protocol/openid-connect/token" 1 strimzi.authorization.client.id="kafka" 2 strimzi.authorization.kafka.cluster.name="kafka-cluster" 1 strimzi.authorization.delegate.to.kafka.acl="true" 1 strimzi.authorization.ssl.truststore.location=<path_to_truststore> 1 strimzi.authorization.ssl.truststore.password=<my_truststore_password> 2 strimzi.authorization.ssl.truststore.type=JKS 3 strimzi.authorization.ssl.secure.random.implementation=SHA1PRNG 4 strimzi.authorization.ssl.endpoint.identification.algorithm=HTTPS 5 strimzi.authorization.grants.refresh.period.seconds="120" 1 strimzi.authorization.grants.refresh.pool.size="10" 2 strimzi.authorization.grants.max.idle.time.seconds="300" 3 strimzi.authorization.grants.gc.period.seconds="300" 4 strimzi.authorization.reuse.grants="false" 5 strimzi.authorization.connect.timeout.seconds="60" 1 strimzi.authorization.read.timeout.seconds="60" 2 strimzi.authorization.http.retries="2" 3 oauth.enable.metrics="true" 1 oauth.include.accept.header="false" 1 authorizer.class.name: com.bisnode.kafka.authorization.OpaAuthorizer opa.authorizer.url=https:// OPA-ADDRESS /allow 1 opa.authorizer.allow.on.error=false 2 opa.authorizer.cache.initial.capacity=50000 3 opa.authorizer.cache.maximum.size=50000 4 opa.authorizer.cache.expire.after.seconds=600000 5 super.users=User:alice;User:bob 6 auto.create.topics.enable=false delete.topic.enable=false ./bin/kafka-topics.sh --bootstrap-server <broker_address> --create --topic <topic_name> --partitions <number_of_partitions> --replication-factor <replication_factor> --config <option_1>=<value_1> --config <option_2>=<value_2> [source,shell,subs=+quotes] ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic mytopic --partitions 50 --replication-factor 3 --config cleanup.policy=compact --config min.insync.replicas=2 ./bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <topic_name> ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic ./bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --describe --topic <topic_name> ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic ./bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --describe ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --describe ./bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --alter --add-config <option>=<value> ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --add-config min.insync.replicas=1 ./bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --alter --delete-config <option> ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --delete-config min.insync.replicas ./bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --delete --topic <topic_name> ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic mytopic ./bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --list bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092 ./bin/connect-standalone.sh ./config/connect-standalone.properties connector1.properties [connector2.properties ...] jcmd | grep ConnectStandalone bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092 group.id=my-group-id config.storage.topic=my-group-id-configs offset.storage.topic=my-group-id-offsets status.storage.topic=my-group-id-status ./bin/connect-distributed.sh ./config/connect-distributed.properties jcmd | grep ConnectDistributed { "name": "my-connector", "config": { "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector", "tasks.max": "1", "topics": "my-topic-1,my-topic-2", "file": "/tmp/output-file.txt" } } curl -X POST -H "Content-Type: application/json" --data @sink-connector.json http://connect0.my-domain.com:8083/connectors curl http://connect0.my-domain.com:8083/connectors curl http://connect0.my-domain.com:8083/connectors curl -X DELETE http://connect0.my-domain.com:8083/connectors/my-connector curl http://connect0.my-domain.com:8083/connectors CLASSPATH=/opt/kafka/libs/connect-file-<kafka_version>.redhat-<build>.jar opt/kafka/bin/connect-distributed.sh plugin.path=/opt/kafka/connector-plugins,/opt/connectors clusters=cluster-1,cluster-2 cluster-1.consumer.fetch.max.bytes=52428800 cluster-2.producer.batch.size=327680 cluster-2.producer.linger.ms=100 cluster-2.producer.request.timeout.ms=30000 clusters=cluster-1,cluster-2 tasks.max = 10 clusters=cluster-1,cluster-2 cluster-1->cluster-2.topics = topic-1, topic-2 cluster-1->cluster-2.groups = group-1, group-2 clusters=cluster-1,cluster-2 1 cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>:443 2 cluster-1.security.protocol=SSL 3 cluster-1.ssl.truststore.password=<truststore_name> cluster-1.ssl.truststore.location=<path_to_truststore>/truststore.cluster-1.jks_ cluster-1.ssl.keystore.password=<keystore_name> cluster-1.ssl.keystore.location=<path_to_keystore>/user.cluster-1.p12 cluster-2.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_two>:443 4 cluster-2.security.protocol=SSL 5 cluster-2.ssl.truststore.password=<truststore_name> cluster-2.ssl.truststore.location=<path_to_truststore>/truststore.cluster-2.jks_ cluster-2.ssl.keystore.password=<keystore_name> cluster-2.ssl.keystore.location=<path_to_keystore>/user.cluster-2.p12 cluster-1->cluster-2.enabled=true 6 cluster-2->cluster-1.enabled=true 7 cluster-1->cluster-2.topics=.* 8 cluster-2->cluster-1.topics=topic-1, topic-2 9 cluster-1->cluster-2.groups=.* 10 cluster-2->cluster-1.groups=group-1, group-2 11 replication.policy.separator=- 12 sync.topic.acls.enabled=false 13 refresh.topics.interval.seconds=60 14 refresh.groups.interval.seconds=60 15 replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy refresh.groups.interval.seconds=60 sync.group.offsets.enabled=true 1 sync.group.offsets.interval.seconds=60 2 emit.checkpoints.interval.seconds=60 3 ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties ./bin/kafka-server-start.sh -daemon ./config/server.properties ./bin/connect-mirror-maker.sh ./config/connect-mirror-maker.properties ./bin/kafka-topics.sh --bootstrap-server <broker_address> --list replication.policy.class=org.apache.kafka.mirror.LegacyReplicationPolicy 1 refresh.topics.enabled=false 2 refresh.groups.enabled=false emit.checkpoints.enabled=false emit.heartbeats.enabled=false sync.topic.configs.enabled=false sync.topic.acls.enabled=false ./bin/kafka-mirror-maker.sh --consumer.config ./config/consumer.properties --producer.config ./config/producer.properties --num.streams=2 ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties ./bin/kafka-server-start.sh -daemon ./config/server.properties ./bin/kafka-topics.sh --bootstrap-server <broker_address> --list export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/my/path/to/log4j.properties"; ./bin/kafka-server-start.sh ./config/server.properties export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/my/path/to/log4j.properties"; ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/my/path/to/connect-log4j.properties"; ./bin/connect-distributed.sh ./config/connect-distributed.properties export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:/my/path/to/connect-log4j.properties"; ./bin/connect-mirror-maker.sh ./config/connect-mirror-maker.properties ./bin/kafka-configs.sh --bootstrap-server <broker_address> --describe --entity-type broker-loggers --entity-name <broker_id> ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type broker-loggers --entity-name 0 # kafka.controller.ControllerChannelManager=INFO sensitive=false synonyms={} kafka.log.TimeIndex=INFO sensitive=false synonyms={} ./bin/kafka-configs.sh --bootstrap-server <broker_address> --alter --add-config " LOGGER-ONE=NEW-LEVEL , LOGGER-TWO=NEW-LEVEL " --entity-type broker-loggers --entity-name <broker_id> ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "kafka.controller.ControllerChannelManager=WARN,kafka.log.TimeIndex=WARN" --entity-type broker-loggers --entity-name 0 Completed updating config for broker: 0. ./bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --delete-config " LOGGER-ONE , LOGGER-TWO " --entity-type broker-loggers --entity-name <broker_id> admin.listeners=https://localhost:8083 admin.listeners.https.ssl.truststore.location=/path/to/truststore.jks admin.listeners.https.ssl.truststore.password=123456 admin.listeners.https.ssl.keystore.location=/path/to/keystore.jks admin.listeners.https.ssl.keystore.password=123456 admin.listeners= cat ./config/connect-log4j.properties log4j.rootLogger=INFO, stdout, connectAppender log4j.logger.org.reflections=ERROR curl -s http://localhost:8083/admin/loggers/ | jq { "org.reflections": { "level": "ERROR" }, "root": { "level": "INFO" } } curl -s http://localhost:8083/admin/loggers/org.apache.kafka.connect.mirror.MirrorCheckpointConnector | jq { "level": "INFO" } curl -Ss -X PUT -H 'Content-Type: application/json' -d '{"level": "TRACE"}' http://localhost:8083/admin/loggers/root { # "org.reflections": { "level": "TRACE" }, "org.reflections.Reflections": { "level": "TRACE" }, "root": { "level": "TRACE" } } client.quota.callback.class=io.strimzi.kafka.quotas.StaticQuotaCallback 1 client.quota.callback.static.produce=1000000 2 client.quota.callback.static.fetch=1000000 3 client.quota.callback.static.storage.per.volume.limit.min.available.bytes=500000000000 4 client.quota.callback.static.storage.check-interval=5 5 client.quota.callback.static.kafka.admin.bootstrap.servers=localhost:9092 6 client.quota.callback.static.excluded.principal.name.list=User:my-user-1;User:my-user-2 7 ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties jcmd | grep Kafka default.replication.factor = 3 min.insync.replicas = 2 ./bin/zookeeper-shell.sh <ip-address>:<zk-port> reconfig -add server.<positive-id> = <address1>:<port1>:<port2>[:role];[<client-port-address>:]<client-port> reconfig -add server.4=172.17.0.4:2888:3888:participant;172.17.0.4:2181 reconfig -remove 5 default.goals=com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.PotentialNwOutGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuUsageDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.TopicReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderReplicaDistributionGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.LeaderBytesInDistributionGoal hard.goals=com.linkedin.kafka.cruisecontrol.analyzer.goals.RackAwareGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.ReplicaCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.DiskCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkInboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.NetworkOutboundCapacityGoal,com.linkedin.kafka.cruisecontrol.analyzer.goals.CpuCapacityGoal curl -v -X POST 'http://<cc_host>:<cc_port>/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal,ReplicaDistributionGoal&skip_hard_goal_check=true' curl -v -X POST 'http://<cc_host>:<cc_port>/kafkacruisecontrol/rebalance' Optimization has 55 inter-broker replica (12 MB) moves, 0 intra-broker replica (0 MB) moves and 24 leadership moves with a cluster model of 5 recent windows and 100.000% of the partitions covered. Excluded Topics: []. Excluded Brokers For Leadership: []. Excluded Brokers For Replica Move: []. Counts: 3 brokers 343 replicas 7 topics. On-demand Balancedness Score Before (78.012) After (82.912). Provision Status: RIGHT_SIZED. a4f833bd-2055-4213-bfdd-ad21f95bf184 sudo mkdir /opt/cruise-control unzip amq-streams-<version>-cruise-control-bin.zip -d /opt/cruise-control sudo chown -R kafka:kafka /opt/cruise-control ./bin/kafka-server-stop.sh metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter cruise.control.metrics.topic.auto.create=true cruise.control.metrics.topic.num.partitions=1 cruise.control.metrics.topic.replication.factor=1 ./bin/kafka-server-start.sh -daemon ./config/server.properties The Kafka cluster to control. bootstrap.servers=localhost:9092 1 The replication factor of Kafka metric sample store topic sample.store.topic.replication.factor=2 2 The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities) #capacity.config.file=config/capacity.json #capacity.config.file=config/capacityCores.json capacity.config.file=config/capacityJBOD.json 3 The list of goals to optimize the Kafka cluster for with pre-computed proposals default.goals={List of default optimization goals} 4 The list of supported goals goals={list of supported optimization goals} 5 The list of supported hard goals hard.goals={List of hard goals} 6 How often should the cached proposal be expired and recalculated if necessary proposal.expiration.ms=60000 7 The zookeeper connect of the Kafka cluster zookeeper.connect=localhost:2181 8 cd ./cruise-control/ ./kafka-cruise-control-start.sh config/cruisecontrol.properties <port_number> curl -X GET 'http://<cc_host>:<cc_port>/kafkacruisecontrol/state' capacity.config.file=config/capacityJBOD.json { "brokerCapacities":[ { "brokerId": "-1", "capacity": { "DISK": "100000", "CPU": "100", "NW_IN": "10000", "NW_OUT": "10000" }, "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB." }, { "brokerId": "0", "capacity": { "DISK": "500000", "CPU": "100", "NW_IN": "50000", "NW_OUT": "50000" }, "doc": "This overrides the capacity for broker 0." } ] } opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name __CruiseControlMetrics --describe ./bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name __CruiseControlMetrics --alter --add-config cleanup.policy=delete curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal,ReplicaDistributionGoal&skip_hard_goal_check=true' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/add_broker?brokerid=3,4' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/remove_broker?brokerid=3,4' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance' * Connected to cruise-control-server (::1) port 9090 (#0) > POST /kafkacruisecontrol/rebalance HTTP/1.1 > Host: cc-host:9090 > User-Agent: curl/7.70.0 > Accept: / > * Mark bundle as not supporting multiuse < HTTP/1.1 200 OK < Date: Mon, 01 Jun 2023 15:19:26 GMT < Set-Cookie: JSESSIONID=node01wk6vjzjj12go13m81o7no5p7h9.node0; Path=/ < Expires: Thu, 01 Jan 1970 00:00:00 GMT < User-Task-ID: 274b8095-d739-4840-85b9-f4cfaaf5c201 < Content-Type: text/plain;charset=utf-8 < Cruise-Control-Version: 2.0.103.redhat-00002 < Cruise-Control-Commit_Id: 58975c9d5d0a78dd33cd67d4bcb497c9fd42ae7c < Content-Length: 12368 < Server: Jetty(9.4.26.v20200117-redhat-00001) curl -v -X POST -H 'User-Task-ID: 274b8095-d739-4840-85b9-f4cfaaf5c201' 'cruise-control-server:9090/kafkacruisecontrol/rebalance' curl -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?dryrun=false' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/add_broker?dryrun=false&brokerid=3,4' curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/remove_broker?dryrun=false&brokerid=3,4' curl 'cruise-control-server:9090/kafkacruisecontrol/user_tasks' USER TASK ID CLIENT ADDRESS START TIME STATUS REQUEST URL c459316f-9eb5-482f-9d2d-97b5a4cd294d 0:0:0:0:0:0:0:1 2020-06-01_16:10:29 UTC Active POST /kafkacruisecontrol/rebalance?dryrun=false 445e2fc3-6531-4243-b0a6-36ef7c5059b4 0:0:0:0:0:0:0:1 2020-06-01_14:21:26 UTC Completed GET /kafkacruisecontrol/state?json=true 05c37737-16d1-4e33-8e2b-800dee9f1b01 0:0:0:0:0:0:0:1 2020-06-01_14:36:11 UTC Completed GET /kafkacruisecontrol/state?json=true aebae987-985d-4871-8cfb-6134ecd504ab 0:0:0:0:0:0:0:1 2020-06-01_16:10:04 UTC curl 'cruise-control-server:9090/kafkacruisecontrol/user_tasks?user_task_ids=c459316f-9eb5-482f-9d2d-97b5a4cd294d' ls -l <LogDir> | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-deleteUSD' ./bin/kafka-server-stop.sh jcmd | grep kafka curl -X POST 'cruise-control-server:9090/kafkacruisecontrol/stop_proposal_execution' cd ./cruise-control/ ./kafka-cruise-control-start.sh config/cruisecontrol.properties <port_number> curl -X GET 'http://<cc_host>:<cc_port>/kafkacruisecontrol/state' ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic <topic_name> --describe curl -X POST 'http://<cc_host>:<cc_port>/kafkacruisecontrol/topic_configuration?topic=<topic_name>&replication_factor=<new_replication_factor>&dryrun=false' { "version": 1, 1 "partitions": [ 2 { "topic": "example-topic-1", 3 "partition": 0, 4 "replicas": [1, 2, 3] 5 }, { "topic": "example-topic-1", "partition": 1, "replicas": [2, 3, 4] }, { "topic": "example-topic-2", "partition": 0, "replicas": [3, 4, 5] } ] } { "version": 1, "topics": [ { "topic": "my-topic"} ] } { "version": 1, "partitions": [ { "topic": "example-topic-1", "partition": 0, "replicas": [1, 2, 3] "log_dirs": ["/var/lib/kafka/data-0/kafka-log1", "any", "/var/lib/kafka/data-1/kafka-log2"] }, { "topic": "example-topic-1", "partition": 1, "replicas": [2, 3, 4] "log_dirs": ["any", "/var/lib/kafka/data-2/kafka-log3", "/var/lib/kafka/data-3/kafka-log4"] }, { "topic": "example-topic-2", "partition": 0, "replicas": [3, 4, 5] "log_dirs": ["/var/lib/kafka/data-4/kafka-log5", "any", "/var/lib/kafka/data-5/kafka-log6"] } ] } { "version": 1, "topics": [ { "topic": "my-topic"} ] } ./bin/kafka-server-start.sh -daemon ./config/server.properties jcmd | grep Kafka ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics.json \ 1 --broker-list 0,1,2,3,4 \ 2 --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3,4],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4,0],"log_dirs":["any","any","any","any"]}]} ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --execute ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --throttle 5000000 --execute ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --verify { "version": 1, "topics": [ { "topic": "my-topic"} ] } ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics.json \ 1 --broker-list 0,1,2,3 \ 2 --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[3,4,2,0],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[0,2,3,1],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[1,3,0,4],"log_dirs":["any","any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]} ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --execute ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --throttle 5000000 --execute ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --verify ls -l <LogDir> | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-deleteUSD' ./bin/kafka-server-stop.sh jcmd | grep kafka { "version": 1, "topics": [ { "topic": "my-topic"} ] } ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics.json \ 1 --broker-list 0,1,2,3,4 \ 2 --generate Current partition replica assignment {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[3,4,2,0],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[0,2,3,1],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[1,3,0,4],"log_dirs":["any","any","any","any"]}]} Proposed partition reassignment configuration {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3,4],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4,0],"log_dirs":["any","any","any","any"]}]} jq '.partitions[].replicas |= del(.[-1])' reassignment.json > reassignment.json {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4],"log_dirs":["any","any","any","any"]}]} ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --execute ./bin/kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassignment.json --verify ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe my-topic Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2 my-topic Partition: 1 Leader: 2 Replicas: 1,2,3 Isr: 1,2,3 my-topic Partition: 2 Leader: 3 Replicas: 2,3,4 Isr: 2,3,4 producer.interceptor.classes=io.opentelemetry.instrumentation.kafkaclients.TracingProducerInterceptor consumer.interceptor.classes=io.opentelemetry.instrumentation.kafkaclients.TracingConsumerInterceptor ./bin/connect-standalone.sh ./config/connect-standalone.properties connector1.properties [connector2.properties ...] ./bin/connect-distributed.sh ./config/connect-distributed.properties header.converter=org.apache.kafka.connect.converters.ByteArrayConverter producer.interceptor.classes=io.opentelemetry.instrumentation.kafkaclients.TracingProducerInterceptor consumer.interceptor.classes=io.opentelemetry.instrumentation.kafkaclients.TracingConsumerInterceptor ./bin/connect-mirror-maker.sh ./config/connect-mirror-maker.properties producer.interceptor.classes=io.opentelemetry.instrumentation.kafkaclients.TracingProducerInterceptor consumer.interceptor.classes=io.opentelemetry.instrumentation.kafkaclients.TracingConsumerInterceptor ./bin/kafka-mirror-maker.sh --producer.config ./config/producer.properties --consumer.config ./config/consumer.properties --num.streams=2 <dependency> <groupId>io.opentelemetry.semconv</groupId> <artifactId>opentelemetry-semconv</artifactId> <version>1.21.0-alpha-redhat-00001</version> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-exporter-otlp</artifactId> <version>1.34.1</version> <exclusions> <exclusion> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-exporter-sender-okhttp</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-exporter-sender-grpc-managed-channel</artifactId> <version>1.34.1</version> <scope>runtime</scope> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId> <version>1.34.1</version> </dependency> <dependency> <groupId>io.opentelemetry.instrumentation</groupId> <artifactId>opentelemetry-kafka-clients-2.6</artifactId> <version>1.32.0-alpha</version> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-sdk</artifactId> <version>1.34.1</version> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-exporter-sender-jdk</artifactId> <version>1.34.1-alpha</version> <scope>runtime</scope> </dependency> <dependency> <groupId>io.grpc</groupId> <artifactId>grpc-netty-shaded</artifactId> <version>1.61.0</version> </dependency> OpenTelemetry ot = GlobalOpenTelemetry.get(); GlobalTracer.register(tracer); // Producer instance Producer < String, String > op = new KafkaProducer < > ( configs, new StringSerializer(), new StringSerializer() ); Producer < String, String > producer = tracing.wrap(op); KafkaTracing tracing = KafkaTracing.create(GlobalOpenTelemetry.get()); producer.send(...); //consumer instance Consumer<String, String> oc = new KafkaConsumer<>( configs, new StringDeserializer(), new StringDeserializer() ); Consumer<String, String> consumer = tracing.wrap(oc); consumer.subscribe(Collections.singleton("mytopic")); ConsumerRecords<Integer, String> records = consumer.poll(1000); ConsumerRecord<Integer, String> record = SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer); senderProps.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName()); KafkaProducer<Integer, String> producer = new KafkaProducer<>(senderProps); producer.send(...); consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName()); KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps); consumer.subscribe(Collections.singletonList("messages")); ConsumerRecords<Integer, String> records = consumer.poll(1000); ConsumerRecord<Integer, String> record = SpanContext spanContext = TracingKafkaUtils.extractSpanContext(record.headers(), tracer); private class TracingKafkaClientSupplier extends DefaultKafkaClientSupplier { @Override public Producer<byte[], byte[]> getProducer(Map<String, Object> config) { KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get()); return telemetry.wrap(super.getProducer(config)); } @Override public Consumer<byte[], byte[]> getConsumer(Map<String, Object> config) { KafkaTelemetry telemetry = KafkaTelemetry.create(GlobalOpenTelemetry.get()); return telemetry.wrap(super.getConsumer(config)); } @Override public Consumer<byte[], byte[]> getRestoreConsumer(Map<String, Object> config) { return this.getConsumer(config); } @Override public Consumer<byte[], byte[]> getGlobalConsumer(Map<String, Object> config) { return this.getConsumer(config); } } KafkaClientSupplier supplier = new TracingKafkaClientSupplier(tracer); KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(config), supplier); streams.start(); props.put(StreamsConfig.PRODUCER_PREFIX + ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName()); props.put(StreamsConfig.CONSUMER_PREFIX + ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName()); OTEL_SERVICE_NAME=my-tracing-service OTEL_TRACES_EXPORTER=zipkin 1 OTEL_EXPORTER_ZIPKIN_ENDPOINT=http://localhost:9411/api/v2/spans 2 //Defines attribute extraction for a producer private static class ProducerAttribExtractor implements AttributesExtractor < ProducerRecord < ? , ? > , Void > { @Override public void onStart(AttributesBuilder attributes, ProducerRecord < ? , ? > producerRecord) { set(attributes, AttributeKey.stringKey("prod_start"), "prod1"); } @Override public void onEnd(AttributesBuilder attributes, ProducerRecord < ? , ? > producerRecord, @Nullable Void unused, @Nullable Throwable error) { set(attributes, AttributeKey.stringKey("prod_end"), "prod2"); } } //Defines attribute extraction for a consumer private static class ConsumerAttribExtractor implements AttributesExtractor < ConsumerRecord < ? , ? > , Void > { @Override public void onStart(AttributesBuilder attributes, ConsumerRecord < ? , ? > producerRecord) { set(attributes, AttributeKey.stringKey("con_start"), "con1"); } @Override public void onEnd(AttributesBuilder attributes, ConsumerRecord < ? , ? > producerRecord, @Nullable Void unused, @Nullable Throwable error) { set(attributes, AttributeKey.stringKey("con_end"), "con2"); } } //Extracts the attributes public static void main(String[] args) throws Exception { Map < String, Object > configs = new HashMap < > (Collections.singletonMap(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")); System.setProperty("otel.traces.exporter", "jaeger"); System.setProperty("otel.service.name", "myapp1"); KafkaTracing tracing = KafkaTracing.newBuilder(GlobalOpenTelemetry.get()) .addProducerAttributesExtractors(new ProducerAttribExtractor()) .addConsumerAttributesExtractors(new ConsumerAttribExtractor()) .build(); dnf install kafka_exporter dnf info kafka_exporter kafka_exporter --kafka.server=<kafka_bootstrap_address>:9092 --kafka.version=3.9.0 --<my_other_parameters> dnf check-update dnf update kafka_exporter mkdir /tmp/kafka unzip amq-streams-<version>-kafka-bin.zip -d /tmp/kafka ./bin/zookeeper-server-stop.sh ./bin/kafka-server-stop.sh jcmd | grep zookeeper jcmd | grep kafka rm -rf /opt/kafka/libs /opt/kafka/bin cp -r /tmp/kafka/kafka_<version>/libs /opt/kafka/ cp -r /tmp/kafka/kafka_<version>/bin /opt/kafka/ rm -r /tmp/kafka inter.broker.protocol.version=3.8 log.message.format.version=3.8 ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties ./bin/kafka-server-start.sh -daemon ./config/server.properties inter.broker.protocol.version=3.9 ./bin/kafka-server-stop.sh jcmd | grep kafka ./bin/kafka-server-start.sh -daemon ./config/server.properties jcmd | grep kafka log.message.format.version=3.9 ./bin/kafka-server-stop.sh jcmd | grep kafka ./bin/kafka-server-start.sh -daemon ./config/server.properties jcmd | grep kafka mkdir /tmp/kafka unzip amq-streams-<version>-kafka-bin.zip -d /tmp/kafka rm -rf ./libs ./bin cp -r /tmp/kafka/kafka_<version>/libs /opt/kafka/ cp -r /tmp/kafka/kafka_<version>/bin /opt/kafka/ rm -r /tmp/kafka ./bin/connect-standalone.sh ./config/connect-standalone.properties <connector1> .properties [ <connector2> .properties ...] ./bin/connect-distributed.sh ./config/connect-distributed.properties ./bin/connect-mirror-maker.sh ./config/connect-mirror-maker.properties ./bin/kafka_bridge_run.sh --config-file= <path> /application.properties jcmd | grep ConnectStandalone jcmd | grep ConnectDistributed jcmd | grep mirrorMaker HTTP-Kafka Bridge started and listening on port 8080 HTTP-Kafka Bridge bootstrap servers localhost:9092 export KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.port=<port> -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false metric.reporters=org.apache.kafka.common.metrics.JmxReporter export KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote=false auto.include.jmx.reporter=false <metric_group>:type=<type_name>,name=<metric_name><other_attribute>=<value> kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec kafka.consumer:type=consumer-fetch-manager-metrics,client-id=<client_id> kafka.consumer:type=consumer-fetch-manager-metrics,client-id=<client_id>,topic=<topic_id> bin/kafka-topics.sh --bootstrap-server :9092 --describe --under-replicated-partitions Topic: topic-1 Partition: 0 Leader: 4 Replicas: 4,2,3 Isr: 4,3 Topic: topic-1 Partition: 1 Leader: 3 Replicas: 2,3,4 Isr: 3,4 Topic: topic-1 Partition: 2 Leader: 3 Replicas: 3,4,2 Isr: 3,4 kafka.server:type=ReplicaManager,name=PartitionCount 1 kafka.server:type=ReplicaManager,name=LeaderCount 2 kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec 3 kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec 4 kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions 5 kafka.server:type=ReplicaManager,name=UnderMinIsrPartitionCount 6 kafka.server:type=KafkaRequestHandlerPool,name=RequestHandlerAvgIdlePercent 1 attributes: OneMinuteRate, FifteenMinuteRate kafka.server:type=socket-server-metrics,listener=([-.\w]+),networkProcessor=([\d]+) 2 attributes: connection-creation-rate kafka.network:type=RequestChannel,name=RequestQueueSize 3 kafka.network:type=RequestChannel,name=ResponseQueueSize 4 kafka.network:type=Processor,name=IdlePercent,networkProcessor=([-.\w]+) 5 kafka.server:type=KafkaServer,name=TotalDiskReadBytes 6 kafka.server:type=KafkaServer,name=TotalDiskWriteBytes 7 kafka.controller:type=KafkaController,name=ActiveControllerCount 1 kafka.controller:type=KafkaController,name=OfflinePartitionsCount 2 kafka.controller:type=ControllerEventManager,name=EventQueueSize 3 requests: EndTxn, Fetch, FetchConsumer, FetchFollower, FindCoordinator, Heartbeat, InitProducerId, JoinGroup, LeaderAndIsr, LeaveGroup, Metadata, Produce, SyncGroup, UpdateMetadata 1 kafka.network:type=RequestMetrics,name=RequestsPerSec,request=([\w]+) 2 kafka.network:type=RequestMetrics,name=RequestQueueTimeMs,request=([\w]+) 3 kafka.network:type=RequestMetrics,name=TotalTimeMs,request=([\w]+) 4 kafka.network:type=RequestMetrics,name=LocalTimeMs,request=([\w]+) 5 kafka.network:type=RequestMetrics,name=RemoteTimeMs,request=([\w]+) 6 kafka.network:type=RequestMetrics,name=ThrottleTimeMs,request=([\w]+) 7 kafka.network:type=RequestMetrics,name=ResponseQueueTimeMs,request=([\w]+) 8 kafka.network:type=RequestMetrics,name=ResponseSendTimeMs,request=([\w]+) 9 attributes: Count, 99thPercentile 10 kafka.consumer:type=consumer-metrics,client-id=([-.\w]+) 1 attributes: time-between-poll-avg, time-between-poll-max kafka.consumer:type=consumer-coordinator-metrics,client-id=([-.\w]+) 2 attributes: heartbeat-response-time-max, heartbeat-rate, join-time-max, join-rate, rebalance-rate-per-hour kafka.producer:type=producer-metrics,client-id=([-.\w]+) 3 attributes: buffer-available-bytes, bufferpool-wait-time, request-latency-max, requests-in-flight attributes: txn-init-time-ns-total, txn-begin-time-ns-total, txn-send-offsets-time-ns-total, txn-commit-time-ns-total, txn-abort-time-ns-total attributes: record-error-total, record-queue-time-avg, record-queue-time-max, record-retry-rate, record-retry-total, record-send-rate, record-send-total #Topic metrics kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic=([-.\w]+) 1 kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic=([-.\w]+) 2 kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec,topic=([-.\w]+) 3 kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec,topic=([-.\w]+) 4 kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=([-.\w]+) 5 kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec,topic=([-.\w]+) 6 kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec,topic=([-.\w]+) 7 #Partition metrics kafka.log:type=Log,name=Size,topic=([-.\w]+),partition=([\d]+)) 8 kafka.log:type=Log,name=NumLogSegments,topic=([-.\w]+),partition=([\d]+)) 9 kafka.log:type=Log,name=LogEndOffset,topic=([-.\w]+),partition=([\d]+)) 10 kafka.log:type=Log,name=LogStartOffset,topic=([-.\w]+),partition=([\d]+)) 11 dnf install <package_name> dnf install <path_to_download_package>