Preface

This document is the user manual for the Kingdee Apusic Distributed Message Queue for Kafka (hereinafter referred to as ADMQ for Kafka) message engine. It provides a detailed introduction to the functional usage, configuration methods, and management operations of the ADMQ for Kafka message engine.

Target Audience

This document is intended for IT business leaders, R&D managers, software project managers, software architects, and operations engineers.

Related Documents

For more information about ADMQ for Kafka, please refer to the following ADMQ for Kafka documentation set:

No.	Document	Description
1	ADMQ for Kafka Quick Start Guide	Briefly introduces how to quickly get started with ADMQ for Kafka.
2	ADMQ for Kafka Installation Guide	Details how to install ADMQ for Kafka on various operating systems, as well as service start/stop operations.
3	ADMQ for Kafka Message Engine User Manual	Details the usage, configuration, management, and accompanying tools of the ADMQ for Kafka message engine.
4	ADMQ for Kafka Console User Manual	Details the functional usage and operation instructions of the ADMQ for Kafka console.
5	ADMQ for Kafka Development Guide	Details client application development for ADMQ for Kafka using various programming languages.
6	ADMQ for Kafka Migration Guide	Details the process of migrating from Kafka to ADMQ for Kafka.
7	ADMQ for Kafka Operations Guide	Details monitoring, operations, security hardening, and other operational instructions for ADMQ for Kafka.
8	ADMQ for Kafka Performance Optimization Guide	Details performance tuning for ADMQ for Kafka.

Technical Support

ADMQ for Kafka provides comprehensive technical support services. You can obtain technical support through the following channels:

• Website: www.apusic.com
• Phone: 400-855-5800
• Email: support@apusic.com
• Kingdee Cloud Community: https://vip.kingdee.com/?productId=73&productLineId=14&lang=zh-CN

When requesting technical support, please provide the following information:

1. Your name

2. Company information and contact details

3. Operating system and its version

4. Product version number

5. Detailed logs, screenshots, etc. of the exception or error

Core Engine Directory Structure

Directory	Description
bin	Contains startup, shutdown, and management scripts
config	Contains configuration files (e.g., kafka-standalone.conf)
data	Stores Kafka message data and ADCC for zk data
jdk	Built-in Java runtime environment, no additional system JDK configuration required
logs	Stores runtime logs (Broker logs, ADCC for zk logs, etc.)
service	System service registration related files (optional)

Broker Configuration Guide

Basic Configuration

Core Purpose: Define the Broker's identity, network exposure method, and storage paths.

Parameter	Description	Production Recommendation
broker.id	Unique cluster identifier. Kafka uses this ID to register itself in ADCC for zk.	Must be unique. If the ID changes after restart, the Broker will be treated as a new node, making old data invisible or causing cluster metadata confusion. Typically hardcoded in the configuration file, not dynamically generated.
listeners	Listener addresses. Defines which IP and port the Broker binds to, and which protocol to use. Format: protocol://IP:port.	- Internal deployment: PLAINTEXT://0.0.0.0:9092 (bind all NICs) or specify internal IP. - Multi-NIC/Hybrid cloud: Can configure multiple, e.g., INTERNAL://192.168.1.10:9092,EXTERNAL://10.0.0.5:9093.
advertised.listeners	Advertised addresses. The address returned to clients when they connect to the Broker. Critically important.	- If clients and Broker are not on the same network (e.g., Docker, cloud servers), must configure this to a public IP or domain accessible to clients. - If not configured, defaults to the value of listeners, which may cause client connection failures to internal IPs.
log.dirs	Data storage directories. Path where message log files are stored.	- Highly recommended to use high-performance SSD. - Can configure multiple directories (comma-separated) to utilize multi-disk IO, e.g., /data1/kafka,/data2/kafka.
num.partitions	Default partition count. Default value when creating a Topic without specifying partitions.	Recommended to set to 3~6 (matching or slightly exceeding the number of cluster Brokers) to improve parallel consumption capacity.
auto.create.topics.enable	Auto-create topics. Whether to automatically create a Topic when a producer sends messages to a non-existent Topic.	**Highly recommended to set to **false** in production. Reason: Prevents creating a large number of useless Topics due to code typos, and auto-created Topic parameters (partitions/replicas) often do not meet production standards.

Replication & Reliability

Core Purpose: Ensure data is not lost and service is highly available.

Parameter	Description	Production Recommendation
default.replication.factor	Default replication factor. Default number of replicas when creating a new Topic.	Recommend ≥ 3. Allows up to 2 machines to fail without data loss. Ensure the number of cluster Brokers ≥ this value.
min.insync.replicas	Minimum in-sync replicas. Used with acks=all, indicates the minimum number of replicas that must acknowledge a write for it to be considered successful.	- If set to 1: Success is returned as long as the Leader writes successfully. Leader failure may cause data loss. - Recommended to set to 2 (with 3 replicas): Ensures that even if 1 node fails, at least 1 full replica survives, and writes require 2 nodes to succeed, balancing safety and availability.
offsets.topic.replication.factor	Replication factor for the __consumer_offsets internal Topic.	Must be ≥ 2 (preferably 3). If this Topic fails, all consumers cannot commit offsets, leading to duplicate consumption or consumption stagnation.
replica.lag.time.max.ms	ISR eviction threshold. If a Follower lags behind the Leader beyond this time, it will be removed from the ISR list.	Default 30s. Can be increased appropriately when network jitter is high, but too large a value will slow down failover.
unclean.leader.election.enable	Non-ISR leader election. Allows replicas not in the ISR (with lagging data) to become Leader.	**Must be set to ****false** (default). If set to true: When all ISR replicas fail, a replica with lagging data is forced to become Leader, which will cause data loss. Strictly prohibited in production.
controller.quorum.voters	Controller quorum voters. Only used in KRaft mode (no ADCC for zk).	Leave empty in ADCC for zk mode. If migrating to KRaft mode, configure the Controller node IDs and addresses.

Log Retention Policy

Core Purpose: Control disk space usage and clean up old data.

Parameter	Description	Production Recommendation
log.retention.hours	Time-based retention. Number of hours to retain log files.	Default 168h (7 days). Adjust based on business requirements, e.g., set to 720 for 30-day compliance retention.
log.retention.bytes	Size-based retention. Maximum bytes to retain per Partition.	Comment this out for default -1 (unlimited). Recommended to set an upper limit (e.g., 50GB) to prevent a single Topic from filling the disk.
log.segment.bytes	Segment file size. Log segment size threshold.	Default 1GB. Too large leads to coarse cleanup granularity; too small leads to excessive file handles. Generally keep the default.
log.retention.check.interval.ms	Check interval. Frequency at which the background thread checks whether to delete old logs.	Default 5 minutes. No need for frequent modification.

Performance Tuning

Core Purpose: Balance throughput and latency, adapt to hardware resources.

Parameter	Description	Production Recommendation
num.network.threads	Network threads. Number of threads handling Socket reads/writes and encoding/decoding.	Default 8. Can be increased in high-concurrency scenarios (e.g., tens of thousands of TPS) or matched to CPU cores.
num.io.threads	IO threads. Number of threads handling disk read/write requests.	Default 16. Recommended to set to 2x the number of disks or the number of CPU cores, e.g., 16.
socket.send/receive.buffer.bytes	Socket buffer. Send/receive buffer size.	Default 1024KB. Can be increased in high-throughput scenarios (e.g., 2MB), contingent on OS kernel parameter tuning.
log.flush.interval.messages/ms	Flush strategy. Number of messages or milliseconds before forcing a flush to disk.	Default value is very large (i.e., no active forced flush). Best practice: Rely on OS Page Cache for asynchronous flushing for optimal performance. Only reduce this value when extreme data zero-loss is required and performance degradation is acceptable.

ADCC for zk Connection

Core Purpose: Connect to the metadata management center (ADCC for zk mode only).

Parameter	Description	Production Recommendation
zookeeper.connect	ADCC for zk address list. Format: host:port, multiple nodes separated by commas.	Fill in all ADCC for zk node addresses. Supports chroot paths.
zookeeper.session.timeout.ms	Session timeout. How long after losing heartbeat with ADCC for zk before a Broker is considered offline.	Default 18s-30s. Smaller values speed up failure detection, but network jitter can easily cause false rebalancing; larger values slow down failure recovery.
zookeeper.chroot.path	Chroot path. Store Kafka metadata under a sub-path in ADCC for zk.	Mandatory when sharing ADCC for zk across multiple clusters. For example, /kafka-cluster-01 to avoid metadata conflicts between different Kafka clusters.

Security

Core Purpose: Authentication and encryption (commented sections are examples).

Parameter	Description	Production Recommendation
security.inter.broker.protocol	Inter-broker communication protocol.	Recommended to enable SASL_SSL or at least SASL_PLAINTEXT in production to prevent internal eavesdropping or node spoofing.
sasl.enabled.mechanisms	Authentication mechanisms.	Common options: PLAIN (simple username/password), SCRAM-SHA-256/512 (more secure), GSSAPI (Kerberos).
listener.name...jaas.config	JAAS configuration.	Defines usernames and passwords. Recommended to move sensitive information to a separate JAAS file rather than writing directly in properties.

Others

Parameter	Description	Production Recommendation
delete.topic.enable	Delete topic switch.	**Recommended ****true**. Facilitates operations for cleaning up test data or deprecated business data. If false, the delete command only marks the topic as invalid without releasing disk space.
JMX_PORT	Monitoring port.	Must be configured. Used by Prometheus/JConsole to collect metrics. Set via -D parameter in startup script, and ensure the firewall allows access.
replica.fetch.wait.max.ms	Follower fetch wait time.	Default 500ms. Increasing can reduce empty polling pressure on the Leader from Followers, but increases sync latency. Generally keep the default.

ADCC for zk Configuration Guide

Basic Timing

Core Purpose: Define heartbeat rhythm and cluster fault tolerance time window, directly impacting failure detection speed.

Parameter	Description	Production Recommendation
tickTime	Heartbeat base time (ms). The basic unit for all internal time calculations in ADCC for zk.	Default 2000ms (2s). - Too small: Network jitter can easily cause false node offline detection. - Too large: Slow failure detection. Recommended to keep at 2000, unless the network is extremely poor, then adjust to 3000-4000.
initLimit	Initialization sync limit. Maximum tickTime multiple for a Follower to sync data from the Leader on startup.	Default 10 (i.e., 20s). If data volume is large or disk is slow, increase (e.g., to 20), otherwise Follower startup may time out and fail.
syncLimit	Heartbeat sync limit. Maximum tickTime multiple for a Follower to respond to Leader heartbeats.	Default 5 (i.e., 10s). If no response within this time, the Leader considers the Follower dead and evicts it from the cluster. Can be increased to 8-10 when the network is unstable.

Data Storage

Core Purpose: Control the data persistence strategy, directly impacting IO performance and disk lifespan.

Parameter	Description	Production Recommendation
dataDir	Snapshot data directory. Stores in-memory database snapshots and myid file.	Must be configured on a high-performance SSD. Do not mix with the transaction log directory (dataLogDir) to avoid IO contention.
dataLogDir	Transaction log directory. Stores transaction logs of write operations.	Key performance point: 1. Must be on a separate disk (physically isolated from dataDir). 2. ADCC for zk has high requirements for sequential writes; a dedicated SSD can significantly improve throughput.
preAllocSize	Pre-allocated file size (KB). Pre-allocated size for transaction log files.	Default 65536 (64MB). Set to 131072 (128MB) in config. Purpose: Reduce file system fragmentation and frequent file extension system calls. Large files benefit sequential write performance.
snapCount	Snapshot trigger frequency. Number of transactions processed before generating a snapshot.	Default 100,000. Set to 200,000 in config. Purpose: Reduce snapshot generation frequency and disk IO pressure. However, this increases restart recovery time (more logs to replay). Can be increased appropriately in production.
fsync.warningthresholdms	Fsync warning threshold (ms). Logs a warning when fsync duration exceeds this value.	Used for monitoring disk health. Frequent warnings indicate severe disk IO bottlenecks requiring hardware replacement or load optimization.

Connection Info

Core Purpose: Define client access points and concurrency control.

Parameter	Description	Production Recommendation
clientPort	Client listening port. The port used by Kafka Brokers and business clients to connect to ADCC for zk.	Default 2181. This port must be opened in the firewall.
maxClientCnxns	Max connections per IP. Limits the maximum number of connections from a single IP address.	Default 60. Set to 0 (unlimited) in config. Risk: If a machine initiates a large number of connections abnormally, it could exhaust ADCC for zk resources. Recommended to set a reasonable upper limit (e.g., 200-500) in production to prevent DDoS or infinite code loops.

Cluster Config

Core Purpose: Define cluster topology and node roles.

Parameter	Description	Production Recommendation
server.x=A:B:C:D;E	Cluster node definition. x: MyID (corresponding to myid file). A: IP. B: Port 2888 (Follower-Leader communication). C: Port 3888 (election communication). D: Role (participant/observer). E: Client port (optional, supported in newer versions).	- Odd number of nodes: Cluster node count must be odd (3, 5, 7) to tolerate (N-1)/2 failures. - Observer: To increase read performance without adding voting overhead, set some nodes as observer.
electionPortBindRetry	Election port binding retry count.	Default 3. Exits after retrying if the port is occupied. Generally no modification needed.

Auto Purge

Core Purpose: Prevent disk from being filled by historical snapshots and logs.

Parameter	Description	Production Recommendation
autopurge.snapRetainCount	Snapshot retain count. Number of most recent snapshot files to retain.	Set to 3 in config. Recommendation: Retain at least 3 to have older versions available for rollback in case the latest snapshot is corrupted.
autopurge.purgeInterval	Purge interval (hours). How often to execute the cleanup task.	Set to 6 (hours) in config. Recommendation: Adjust based on data volume. For high write scenarios, set to 1-2 hours; for low load, set to 12-24 hours.

Network Tuning

Core Purpose: Handle high-concurrency connections and session management.

Parameter	Description	Production Recommendation
globalOutstandingLimit	Global outstanding request limit. Prevents request堆积 from causing OOM.	Default 1000. Set to 5000 in config. Note: Increasing can improve throughput, but if backend processing is slow, it may cause memory spikes. Monitor memory usage.
minSessionTimeout	Minimum session timeout (ms). Client request timeouts below this value will be forcibly set to this value.	Default 2 * tickTime. Setting too small can cause disconnection with minor network jitter. Recommend ≥ 4000ms.
maxSessionTimeout	Maximum session timeout (ms).	Default 20 * tickTime. Set to 60000 (60s) in config. Allows long-lived clients to set longer timeouts, reducing reconnection frequency.

Admin & 4lw

Core Purpose: Operations monitoring and security control.

Parameter	Description	Production Recommendation
admin.enableServer	Enable AdminServer (HTTP interface).	Default true (newer versions). Set to false in config. Security recommendation: If HTTP monitoring interface is not needed, recommend disabling to reduce the attack surface. Monitor via JMX or 4lw commands.
4lw.commands.whitelist	Four-letter word command whitelist. ZK monitoring via echo cmd \| nc ip 2181.

Others

Parameter	Description	Production Recommendation
skipACL	Skip ACL check.	Default no. If set to yes, all permission checks are ignored, greatly improving performance but losing security. Use only in fully trusted internal networks when pursuing maximum performance.
leaderServers	Whether the Leader handles read requests.	Config has no (may be an older version parameter or specific distribution). In standard ADCC for zk, the Leader handles all write requests and some read requests by default. If set not to participate, it may affect read throughput. Generally keep the default.
forceSync	Force fsync on every transaction.	Default no (uses OS buffering). If set to yes, data is most secure but performance drops sharply. **Production environment must keep ****no**.
sslQuorum / portUnification	SSL encryption and port unification.	If SSL encryption for cluster communication is needed, set to true and configure certificates. Keep as false for plaintext internal network communication.

Common Commands

Topic Management

Create Topic

Create a Topic named test-topic with 3 partitions and 3 replicas.

kafka/bin/kafka-topics.sh --create \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic \
  --partitions 3 \
  --replication-factor 3

• Note: replication-factor cannot exceed the number of Brokers in the cluster.

List Topics

# List all topics
kafka/bin/kafka-topics.sh --list --bootstrap-server 192.168.1.10:9092

# Filter specific topics (supports regex)
kafka/bin/kafka-topics.sh --list --bootstrap-server 192.168.1.10:9092 | grep "test"

View Topic Details (Partition & Replica Distribution)

This is the most important command for diagnosing data skew and replica status.

kafka/bin/kafka-topics.sh --describe \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic

Output Interpretation:

• Leader: The replica ID currently responsible for reads and writes.
• Replicas: List of Broker IDs where all replicas of this partition reside.
• Isr (In-Sync Replicas): List of replicas currently in sync with the Leader. If the Isr count is less than Replicas, it indicates some replicas are lagging or down.

Increase Partitions (Can only increase, not decrease)

Increase the partition count of test-topic from 3 to 6.

kafka/bin/kafka-topics.sh --alter \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic \
  --partitions 6

Delete Topic

kafka/bin/kafka-topics.sh --delete \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic

• Prerequisite: delete.topic.enable=true in the server.properties configuration file.

---

Produce & Consume (Data Testing)

Start Console Producer

Send messages to test-topic. Enter a line and press Enter to send a message.

kafka/bin/kafka-console-producer.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic

• Advanced Usage: Specify Key for sending (for testing partition strategy)

kafka/bin/kafka-console-producer.sh --bootstrap-server 192.168.1.10:9092 --topic test-topic --property "parse.key=true" --property "key.separator=:"
# Input format: key:value  (e.g., user1:login)

Start Console Consumer

Consume messages from test-topic.

• Consume latest data:

kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic

• Consume from beginning (historical data):

kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic \
  --from-beginning

• Display Key and Partition information:

kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --topic test-topic \
  --from-beginning \
  --property print.key=true \
  --property partition=true \
  --property print.timestamp=true

• Limit consumption count (for testing):

kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.10:9092 --topic test-topic --max-messages 10

Consumer Group Management

List Consumer Groups

kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --list

View Group Details (Lag, Offset)

The most commonly used operations command for viewing consumption lag.

kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server 192.168.1.10:9092 \
  --group my-consumer-group \
  --describe

Key Output Columns:

• CURRENT-OFFSET: The currently committed position.
• LOG-END-OFFSET: The latest message position in the partition.
• LAG: Backlog = LOG-END-OFFSET - CURRENT-OFFSET. If LAG keeps increasing, consumption speed cannot keep up with production speed.

Reset Offset (Use with Caution)

Reset the consumer group's offset to earliest, latest, or a specific time.

• Reset to earliest:

kafka/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.10:9092 \
  --group my-consumer-group \
  --topic test-topic \
  --reset-offsets --to-earliest \
  --execute

• Reset to latest (ignore historical backlog):

kafka/bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.10:9092 \
  --group my-consumer-group \
  --topic test-topic \
  --reset-offsets --to-latest \
  --execute

• Note: When performing a reset, the consumer group must be in an inactive state (i.e., no running consumer instances).

Cluster & Metadata Management

View Cluster Broker Information

kafka/bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.1.10:9092

Preferred Leader Election

If a failure causes uneven Leader distribution (e.g., all Leaders on Node 1), trigger a re-election to return Leaders to the first node in the Replicas list (usually the ideal distribution).

kafka/bin/kafka-leader-election.sh --bootstrap-server 192.168.1.10:9092 \
  --election-type preferred \
  --all-topic-partitions

View Cluster Configuration

kafka/bin/kafka-configs.sh --bootstrap-server 192.168.1.10:9092 \
  --entity-type brokers \
  --entity-name 0 \
  --describe

Common Issues

Common Error Troubleshooting

• ADCC for zk startup fails with Connection refused or Session expired in logs

  • Check if the myid file exists and has the correct content.
  • Check if the IP in server.x in zookeeper.properties is reachable, and if ports 2888/3888 are blocked by the firewall.

• Kafka startup fails with Broker ID 0 is already registered in logs

  • Check for residual old ADCC for zk data. For a new deployment, clear ADCC for zk's dataDir and Kafka's log.dirs and retry.
  • Confirm that broker.id is not duplicated across nodes.

• Incomplete ISR set (Isr: 0,1 instead of 0,1,2)

  • Indicates a Follower is syncing too slowly or the network is unreachable. Check the node's logs and network bandwidth.

• Kafka startup fails with Failed to bind to port 9092 in logs

  • Check if port 9092 is occupied by another process (use netstat -tulpn | grep 9092 or lsof -i:9092).
  • Check if the port configured in listeners in server.properties matches the intended port to avoid configuration conflicts.
  • Confirm that the firewall is not blocking port 9092, or that the port has been allowed in firewall rules.

• Kafka producer fails to send messages with Leader not available

  • Check if the corresponding Topic exists; if not, create it first (using kafka-topics.sh --create).
  • Check if the Kafka cluster is running normally and all Broker nodes are online.
  • Check if the Topic's replica configuration is reasonable; if the replication factor exceeds the number of available Brokers, Leader election cannot proceed.

• Kafka consumer fails to consume with No offset found for partition

  • For a new consumer group, having no offset on first consumption is normal. Configure auto.offset.reset to earliest or latest.
  • For an existing consumer group, check if offset data in ADCC for zk/Kafka was manually deleted, or if the consumer group has not consumed for a long time causing offsets to expire (adjust the offsets.retention.minutes configuration).

• After ADCC for zk cluster starts, some nodes cannot join the cluster with Cannot open channel to x at election address in logs

  • Check if the IP/port configured in server.x in zoo.cfg is consistent across all ADCC for zk nodes, with no typos.
  • Check if the system clocks are synchronized across nodes (time difference exceeding 2000ms can cause election failure); use ntpdate to sync time.
  • Confirm that the dataDir directory permissions allow the ADCC for zk running user to read and write, preventing the myid file from being unreadable due to insufficient permissions.

• Kafka logs show Not enough replicas, messages cannot be persisted

  • Check if the Topic's min.insync.replicas value is greater than the currently available number of in-sync replicas (ISR).
  • For a single-Broker deployment, set min.insync.replicas to 1 (default is 1; adjust if manually changed).
  • Check if Follower nodes are syncing normally; investigate network latency and high disk IO that may cause slow sync.

• Kafka crashes with JVM out-of-memory (OOM) after startup

  • Check the JVM configuration (KAFKA_HEAP_OPTS) in kafka-server-start.sh; the default heap size may be too large or too small.
  • For standalone test environments, set to -Xmx512M -Xms512M; for production, adjust based on server memory (recommended not to exceed 50% of physical memory).
  • Check if too many Topics/partitions are enabled, causing high memory usage. Plan the number of partitions reasonably.

• SSL handshake failed when connecting to Kafka

  • If SSL authentication is enabled, check if the SSL certificate path and password configuration on the client/server are correct.
  • Confirm that the certificate has not expired and that the server's listeners is configured as SSL:// rather than PLAINTEXT://.
  • Check if the client has correctly configured SSL-related parameters (e.g., security.protocol=SSL).