Development Manual

# Preface

This document provides development instructions for the Kingdee Apusic Distributed Message Queue for Kafka (Apusic Distributed Message Queue, hereinafter referred to as ADMQ for Kafka), helping users quickly learn how to develop using ADMQ for Kafka.

# Target Audience

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

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 Architecture

# Topics and Partitions

  • Topic: Logical classification of messages
  • Partition: Physical shard of a Topic, enabling parallel reading/writing and horizontal scaling
    • Messages within a single partition are strictly ordered
    • No global order guarantee across different partitions
    • The number of partitions should be set based on throughput requirements, typically 10-50

# Replication Mechanism

  • Leader Replica: Handles all read and write requests
  • Follower Replica: Only syncs data from the Leader, does not handle client requests
  • ISR (In-Sync Replicas): The set of replicas that are in sync with the Leader
    • Only replicas in the ISR are eligible to become a new Leader
    • Controlled by replica.lag.time.max.ms (default 30 seconds)

# High Availability and Failover

  • Controller monitors Broker status

  • When a Leader fails, a new Leader is elected from the ISR (the replica with the highest offset is preferred)

  • Cluster metadata is updated, and clients automatically reconnect to the new Leader

  • Non-ISR replicas are prohibited from becoming Leader by default (unclean.leader.election.enable=false)

# Installation and Deployment

# Standalone Deployment

Extract the installation package

mkdir /apusic
tar -zxf ADMQ-V2.0.6.391-Kafka-20260319.tar.gz -C /apusic
1
2

Modify configuration

cd /apusic/admq-kafka
vi config/kafka-standalone.conf
...
# Change the address below to the actual server IP
listeners=PLAINTEXT://192.168.1.10:9092
...
1
2
3
4
5
6

Start

bin/admq-daemon start kafka zk 
bin/admq-daemon start kafka standalone
1
2

# Client Development

# Maven Dependency

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.1</version>
</dependency>
1
2
3
4
5

# Producer Core Configuration

Config Recommended Value Description
bootstrap.servers host1:9092,host2:9092 Cluster address list
acks all Wait for acknowledgement from all ISR replicas
enable.idempotence true Enable idempotence to prevent duplicates
compression.type lz4 or zstd Compression algorithm
linger.ms 5-10 Batch send delay
batch.size 16384 Batch size (bytes)
retries Integer.MAX_VALUE Number of retries
max.in.flight.requests.per.connection 5 Max unacknowledged requests per connection (≤ 5 when idempotence enabled)

# Producer Example Code

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "value1");
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        exception.printStackTrace();
    } else {
        System.out.println("Sent to partition " + metadata.partition() + 
                          ", offset " + metadata.offset());
    }
});

producer.close();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# Consumer Core Configuration

Config Recommended Value Description
bootstrap.servers host1:9092,host2:9092 Cluster address
group.id Custom Consumer group ID
enable.auto.commit false Disable auto-commit, manual control
auto.offset.reset earliest or latest Strategy when no offset exists
max.poll.records 500 Max records per single poll
session.timeout.ms 30000 Session timeout
heartbeat.interval.ms 10000 Heartbeat interval (should be less than 1/3 of session timeout)

# Consumer Example Code

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Received: " + record.value());
        }
        consumer.commitSync();
    }
} finally {
    consumer.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# Production Best Practices

# Topic Design

  • Partition count calculation: Single partition throughput × number of partitions ≥ target throughput
  • Replication factor: 3 replicas recommended for production
  • Cleanup policy:
    • cleanup.policy=delete: Clean by time/size
    • cleanup.policy=compact: Log compaction, retain the latest Key

# Performance Tuning

  • Producer side:
    • Enable compression (lz4 or zstd)
    • Set linger.ms and batch.size appropriately
    • Enable idempotence and transactions (Exactly-Once semantics)
  • Consumer side:
    • Adjust max.poll.records to avoid slow single processing
    • Set fetch.min.bytes and fetch.max.wait.ms appropriately

# Monitoring Metrics

Metric Description Alert Threshold
UnderReplicatedPartitions Number of under-replicated partitions > 0
OfflinePartitionsCount Number of offline partitions > 0
RequestLatencyAvg Average request latency > 100ms
MessagesInPerSec Messages per second Based on business
BytesInPerSec Bytes written per second Based on business

# Security Configuration

# SASL/PLAIN Authentication Configuration

# broker configuration
listeners=SASL_PLAINTEXT://0.0.0.0:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false

# JAAS configuration file (passwords below are examples only; use strong passwords in production)
KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="admin-secret"
   user_admin="admin-secret"
   user_alice="alice-secret";
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# ACL Example

# Grant alice read and write permissions on topic my-topic
kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:alice \
  --operation Read --operation Write \
  --topic my-topic

# Grant alice operation permission on consumer group my-group
kafka-acls.sh --bootstrap-server localhost:9092 \
  --add --allow-principal User:alice \
  --operation Read --group my-group
1
2
3
4
5
6
7
8
9
10

# Common Commands

ADMQ for Kafka provides native commands located in the admq-kafka/kafka/bin/ directory.

# Topic Management

cd /apusic/admq-kafka/kafka/bin/

# Create topic
kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic my-topic --partitions 3 --replication-factor 3

# List all topics
kafka-topics.sh --bootstrap-server localhost:9092 --list

# View topic details
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic

# Modify partition count (can only increase)
kafka-topics.sh --bootstrap-server localhost:9092 --alter \
  --topic my-topic --partitions 6

# Delete topic
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my-topic
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# Consumer Group Management

# List consumer groups
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

# View consumer group details
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group my-group

# Reset offsets
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group --reset-offsets --to-earliest \
  --topic my-topic --execute
1
2
3
4
5
6
7
8
9
10
11

# KRaft Mode Operations

# Format storage directory (first time only)
kafka-storage.sh format -t <cluster-id> -c config/kraft/server.properties

# View quorum status
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

# Dynamically add controller node
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 add \
  --controller-quorum-voter <node-id>@<host>:<port>
1
2
3
4
5
6
7
8
9

# Troubleshooting

# Common Issues

Issue Possible Cause Solution
Frequent consumer rebalancing session.timeout.ms too small or processing time too long Increase session timeout, optimize consumption logic
Message loss acks=1 or idempotence not enabled Set acks=all, enable enable.idempotence
Duplicate consumption Offset commit failure Use manual commit, handle deduplication logic
High producer latency Network or disk bottleneck Check network bandwidth, optimize disk I/O
Partition unavailable Broker failure or insufficient replicas Check Broker status, ensure sufficient ISR

# Log Locations

  • Broker logs: logs/server.log
  • Controller logs: logs/controller.log
  • State change logs: logs/state-change.log
编辑页面 (opens new window)
#开发手册

← Migration Manual→