Kafka 2 — Producer chuyên sâu

Giới thiệu

Producer là thành phần đẩy dữ liệu vào Kafka. Trong bối cảnh ngân hàng, mỗi giao dịch (chuyển khoản, rút tiền, thanh toán thẻ) là một bản ghi cần được ghi vào topic một cách không mất mát và không trùng lặp. Một khoản chuyển tiền bị ghi hai lần, hay bị mất sau khi đã trừ tiền tài khoản nguồn, đều là lỗi nghiêm trọng. Vì vậy hiểu chính xác cách producer hoạt động — từ luồng gửi, cách chọn partition, đến các đảm bảo độ bền (acks, enable.idempotence, transactions) — là điều bắt buộc.

Bài này đi sâu vào cơ chế nội tại của producer và cách cấu hình cho một hệ thống ghi giao dịch tài chính đáng tin cậy.

Luồng gửi của producer

Khi bạn gọi producer.send(record), bản ghi không được gửi ngay lên broker. Nó đi qua một chuỗi bước xử lý bất đồng bộ:

Các bước chi tiết:

1. Serialize: key và value được chuyển thành mảng byte bằng key.serializer và value.serializer. Với giao dịch, thường dùng Avro/Protobuf qua Schema Registry để đảm bảo tương thích schema.
2. Partitioner: quyết định bản ghi đi vào partition nào của topic (xem phần dưới).
3. RecordAccumulator (buffer/batch): bản ghi được gom vào một batch trong bộ nhớ đệm, nhóm theo (topic, partition). Đây là nơi batch.size, linger.ms, buffer.memory phát huy tác dụng.
4. Sender thread: một luồng nền (I/O thread) gom các batch đã sẵn sàng, nhóm chúng theo broker leader, và tạo ProduceRequest.
5. Gửi và nhận phản hồi: broker leader ghi vào log, replicate theo yêu cầu của acks, rồi trả ProduceResponse. Kết quả (RecordMetadata hoặc exception) được trả về qua Future hoặc Callback.

Điểm mấu chốt: send() là bất đồng bộ và trả về Future<RecordMetadata>. Việc gửi thực sự diễn ra ở Sender thread, tách rời khỏi luồng ứng dụng.

Chọn partition

Partition quyết định thứ tự và khả năng song song. Kafka chỉ đảm bảo thứ tự trong một partition, không phải toàn topic.

Có key: hash partitioning

Khi bản ghi có key, partitioner mặc định tính:

partition = murmur2(serialized_key) % numPartitions

Cùng một key → luôn cùng một partition (miễn số partition không đổi). Đây là cơ chế then chốt cho giao dịch ngân hàng: dùng số tài khoản làm key đảm bảo mọi giao dịch của cùng một tài khoản đi vào cùng một partition, giữ đúng thứ tự thời gian.

producer.send("transactions", key=b"ACC-000123", value=payload)

Lưu ý: nếu tăng số partition sau này, ánh xạ key → partition thay đổi, phá vỡ đảm bảo thứ tự lịch sử. Vì vậy với dữ liệu cần thứ tự, hãy chọn số partition đủ lớn ngay từ đầu.

Không key: sticky partitioning

Khi key = null, các phiên bản Kafka hiện đại (từ 2.4 trở đi) dùng sticky partitioner: producer "dính" vào một partition cho đến khi batch hiện tại đầy hoặc linger.ms hết hạn, rồi mới chuyển sang partition khác. Điều này tạo batch lớn hơn, giảm số request, giảm độ trễ đuôi — tốt hơn round-robin thuần từng bản ghi (cách cũ). Kết quả vẫn phân phối tương đối đều giữa các partition theo thời gian.

Custom partitioner

Khi cần logic riêng — ví dụ định tuyến giao dịch VIP vào partition riêng — có thể cài partitioner.class:

public class VipPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionsForTopic(topic).size();
        if (key != null && key.toString().startsWith("VIP-")) {
            return 0; // partition dành riêng cho VIP
        }
        // các key khác: hash trên phần còn lại
        return Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1) + 1;
    }
    @Override public void close() {}
    @Override public void configure(Map<String, ?> configs) {}
}

acks — đánh đổi độ bền và độ trễ

acks quy định producer chờ bao nhiêu bản sao (replica) xác nhận trước khi coi việc gửi là thành công.

Với giao dịch ngân hàng, luôn dùng acks=all. Không được chấp nhận mất một khoản chuyển tiền chỉ vì leader chết ngay sau khi phản hồi.

min.insync.replicas liên hệ với acks=all

acks=all một mình chưa đủ. all nghĩa là chờ tất cả replica trong ISR (In-Sync Replicas). Nếu ISR chỉ còn 1 (các follower đã tụt lại và bị loại), thì acks=all thực chất chỉ như acks=1 — mất bản sao đó là mất dữ liệu.

Cấu hình cấp broker/topic min.insync.replicas đặt số replica đồng bộ tối thiểu. Với min.insync.replicas=2 và acks=all:

• Ghi chỉ thành công khi ít nhất 2 replica đã nhận.
• Nếu ISR tụt xuống dưới 2, broker từ chối ghi (ném NotEnoughReplicasException), thà lỗi còn hơn âm thầm mất dữ liệu.

Công thức an toàn phổ biến: replication.factor=3, min.insync.replicas=2, acks=all. Cho phép mất 1 broker mà vẫn ghi được, và không mất dữ liệu khi mất 1 broker.

Idempotent producer

Retry là con dao hai lưỡi. Giả sử producer gửi giao dịch, broker ghi thành công nhưng ProduceResponse bị mất do lỗi mạng. Producer không nhận được xác nhận nên gửi lại — kết quả: giao dịch bị ghi hai lần. Đây là "at-least-once" gây trùng lặp.

Idempotent producer giải quyết trùng lặp này. Bật bằng:

enable.idempotence=true

Cơ chế:

• Mỗi producer được broker cấp một PID (Producer ID) duy nhất.
• Mỗi bản ghi mang theo một sequence number tăng dần theo từng (PID, partition).
• Broker theo dõi sequence cao nhất đã ghi cho mỗi (PID, partition). Nếu nhận lại một sequence đã thấy (do retry), broker loại bỏ bản trùng và vẫn trả thành công.

Kết quả: exactly-once ghi vào một partition, ngay cả khi có retry. Điều này chỉ áp dụng trong phạm vi một phiên producer và một partition.

Khi bật enable.idempotence=true, Kafka tự động ép các ràng buộc:

• acks=all
• retries > 0 (mặc định Integer.MAX_VALUE)
• max.in.flight.requests.per.connection <= 5

Từ Kafka 3.0, enable.idempotence=true là mặc định. Nhưng nên khai báo tường minh để tránh nhầm lẫn khi triển khai lên các cluster phiên bản khác nhau.

Transactions — exactly-once giữa các topic (giới thiệu)

Idempotence chỉ đảm bảo không trùng trong một partition. Khi cần atomic trên nhiều partition/topic — ví dụ ghi giao dịch vào topic transactions và cập nhật topic account-balances phải cùng thành công hoặc cùng thất bại — ta dùng transactions.

Cấu hình:

transactional.id=txn-writer-app-1

Đặt transactional.id sẽ tự bật idempotence. Luồng dùng:

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("transactions", accountId, txnEvent));
    producer.send(new ProducerRecord<>("account-balances", accountId, newBalance));
    producer.commitTransaction();
} catch (KafkaException e) {
    producer.abortTransaction();
}

Consumer đọc dữ liệu transactional cần đặt isolation.level=read_committed để chỉ thấy các bản ghi đã commit. Đây là nền tảng của mô hình exactly-once processing (đặc biệt trong consume-transform-produce của Kafka Streams). Ta sẽ đi sâu ở bài về stream processing.

Batching, linger và compression

Producer gom nhiều bản ghi thành batch để tăng throughput. Ba tham số chính:

• batch.size (byte, mặc định 16384): kích thước tối đa của một batch cho mỗi partition. Batch đầy → gửi ngay. Tăng giá trị này giúp throughput cao hơn nhưng tốn bộ nhớ và có thể tăng độ trễ.
• linger.ms (mặc định 0): thời gian producer chờ thêm để gom bản ghi trước khi gửi, ngay cả khi batch chưa đầy. linger.ms=0 gửi ngay khi có thể; đặt linger.ms=5..20 cho phép gom batch lớn hơn, đổi chút độ trễ lấy throughput và tỷ lệ nén tốt hơn.
• buffer.memory (mặc định 33554432 = 32MB): tổng bộ nhớ đệm cho tất cả batch chưa gửi. Khi đầy, send() sẽ chặn tối đa max.block.ms rồi ném exception.

Compression giảm băng thông mạng và dung lượng lưu trữ, thường tăng throughput vì batch nén nhỏ hơn:

compression.type=zstd

Các lựa chọn: none, gzip, snappy, lz4, zstd. Kinh nghiệm: lz4 cân bằng tốc độ/tỷ lệ nén tốt; zstd cho tỷ lệ nén cao hơn (tốt cho dữ liệu giao dịch JSON/text lặp lại nhiều) với chi phí CPU cao hơn một chút; snappy nhanh, tỷ lệ nén vừa phải. Compression áp dụng theo cả batch nên linger.ms lớn hơn → batch lớn hơn → nén hiệu quả hơn.

max.in.flight và ảnh hưởng thứ tự

max.in.flight.requests.per.connection là số request chưa được xác nhận mà producer cho phép gửi song song trên một kết nối.

• Giá trị cao (mặc định 5) → throughput cao hơn.
• Nhưng nếu không bật idempotence và có retry: một request phía sau có thể được ghi thành công trước request phía trước bị retry, làm đảo thứ tự trong partition.

Khi enable.idempotence=true, broker dùng sequence number để sắp xếp lại đúng thứ tự ngay cả khi có tối đa 5 request in-flight và có retry. Vì vậy với idempotence bật, bạn vừa giữ được thứ tự vừa giữ throughput cao — đây là lý do nữa để luôn bật idempotence.

Nếu (vì lý do nào đó) không bật idempotence mà vẫn cần thứ tự tuyệt đối, phải đặt max.in.flight.requests.per.connection=1, đánh đổi throughput.

Retries và delivery.timeout.ms

• retries (mặc định Integer.MAX_VALUE khi idempotence bật): số lần thử lại với lỗi tạm thời (leader chưa bầu xong, NotEnoughReplicas, timeout mạng...).
• retry.backoff.ms (mặc định 100): thời gian chờ giữa các lần retry.
• delivery.timeout.ms (mặc định 120000 = 2 phút): giới hạn thời gian tổng từ lúc send() trả về đến khi thành công hoặc thất bại vĩnh viễn. Đây là "chốt chặn" thực sự — bao trùm cả thời gian trong buffer, các lần retry và request.timeout.ms. Ràng buộc: delivery.timeout.ms >= linger.ms + request.timeout.ms.

Trong thực tế, nên điều chỉnh delivery.timeout.ms thay vì dựa vào retries. Với hệ thống giao dịch, đặt timeout đủ dài để vượt qua các sự cố bầu leader ngắn hạn, nhưng đủ ngắn để phát hiện lỗi kịp thời và kích hoạt xử lý dự phòng (ví dụ ghi vào dead-letter store).

Error handling

Lỗi producer chia hai nhóm:

• Retriable (RetriableException): LeaderNotAvailableException, NotEnoughReplicasException, TimeoutException... — producer tự retry cho đến khi hết delivery.timeout.ms.
• Non-retriable / fatal: SerializationException, RecordTooLargeException, ProducerFencedException, AuthorizationException... — thất bại ngay, cần xử lý ở ứng dụng.

Với giao dịch ngân hàng, tuyệt đối không nuốt lỗi âm thầm. Callback phải kiểm tra exception và, nếu thất bại vĩnh viễn, đẩy bản ghi vào một cơ chế dự phòng (dead-letter topic, hàng đợi retry ứng dụng, hoặc alert) thay vì bỏ qua.

Ví dụ cấu hình producer

Cấu hình khuyến nghị cho ghi giao dịch ngân hàng (độ bền cao):

bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=io.confluent.kafka.serializers.KafkaAvroSerializer

# Độ bền + không trùng lặp
acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5

# Ràng buộc phía topic (đặt khi tạo topic): min.insync.replicas=2, replication.factor=3

# Throughput
batch.size=32768
linger.ms=10
compression.type=zstd
buffer.memory=67108864

# Timeout / retry
delivery.timeout.ms=120000
request.timeout.ms=30000
retry.backoff.ms=100
max.block.ms=60000

Ví dụ code gửi

Python — confluent-kafka

from confluent_kafka import Producer
import json

conf = {
    "bootstrap.servers": "kafka1:9092,kafka2:9092,kafka3:9092",
    "acks": "all",
    "enable.idempotence": True,
    "compression.type": "zstd",
    "linger.ms": 10,
    "batch.size": 32768,
    "delivery.timeout.ms": 120000,
}
producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        # KHÔNG nuốt lỗi: đẩy sang dead-letter / alert
        print(f"Ghi thất bại key={msg.key()}: {err}")
    else:
        print(f"OK topic={msg.topic()} partition={msg.partition()} "
              f"offset={msg.offset()}")

txn = {
    "txn_id": "T-20260630-000987",
    "account_id": "ACC-000123",
    "amount": -500000,
    "currency": "VND",
    "type": "TRANSFER_OUT",
    "ts": "2026-06-30T10:15:00+07:00",
}

# key = số tài khoản -> mọi giao dịch của tài khoản này cùng partition, giữ thứ tự
producer.produce(
    topic="transactions",
    key=txn["account_id"].encode("utf-8"),
    value=json.dumps(txn).encode("utf-8"),
    callback=delivery_report,
)

# poll xử lý callback đã hoàn tất; flush đảm bảo mọi bản ghi được gửi trước khi thoát
producer.poll(0)
producer.flush()

Java — KafkaProducer

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());
props.put("acks", "all");
props.put("enable.idempotence", true);
props.put("compression.type", "zstd");
props.put("linger.ms", 10);
props.put("delivery.timeout.ms", 120000);

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("transactions", "ACC-000123", payloadJson);

    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            // xử lý lỗi: retry vĩnh viễn -> dead-letter / alert
            log.error("Ghi giao dịch thất bại", exception);
        } else {
            log.info("OK partition={} offset={}",
                     metadata.partition(), metadata.offset());
        }
    });
}
// try-with-resources tự gọi close() -> flush toàn bộ bản ghi còn trong buffer

Tóm tắt

Producer gửi bản ghi qua chuỗi serialize → partitioner → buffer/batch → sender thread, tất cả bất đồng bộ. key quyết định partition (hash) và do đó thứ tự; không key thì dùng sticky partitioner. acks=all kết hợp min.insync.replicas=2 và replication.factor=3 cho độ bền cao — bắt buộc với giao dịch ngân hàng. enable.idempotence=true (mặc định từ 3.0) dùng PID + sequence để tránh trùng lặp khi retry, đồng thời giữ đúng thứ tự dù max.in.flight tới 5. Transactions với transactional.id mở rộng exactly-once ra nhiều topic. batch.size, linger.ms, compression.type và buffer.memory điều chỉnh throughput; delivery.timeout.ms là chốt chặn thời gian tổng. Error handling phải rõ ràng, không bao giờ nuốt lỗi.

Tự kiểm tra

1. Vì sao acks=all một mình chưa đủ để đảm bảo không mất dữ liệu, và min.insync.replicas bổ sung điều gì?
2. Idempotent producer dùng cơ chế gì để loại bỏ bản ghi trùng khi có retry?
3. Nếu tăng số partition của một topic đang dùng key làm định tuyến, đảm bảo nào bị phá vỡ?
4. Vì sao khi bật enable.idempotence=true, max.in.flight.requests.per.connection=5 vẫn giữ được thứ tự trong partition?
5. Sự khác biệt giữa retries, delivery.timeout.ms và request.timeout.ms là gì?
6. Khi nào cần dùng transactions thay vì chỉ dùng idempotence?

Đọc tiếp

Kafka 3 — Consumer & Consumer Group