Streams: StreamProducer and StreamConsumer

Send records to topics and read them in consumer groups with one API on Amazon MSK, Azure Event Hubs (through its Kafka endpoint) and Google Cloud Managed Service for Apache Kafka, plus in-memory topics. All three are Kafka: records are kept in partitioned, ordered logs, and each consumer group keeps its own position in each partition. Delivery is at least once, so make handlers idempotent.

Topics and consumer group permissions are infrastructure: create them with your usual tooling. Halyx never creates topics.

Streams or messaging? Use messaging for work queues: each message is acknowledged on its own, and a failed one is redelivered later while others carry on. Use streams for event logs: order is kept per key, consumers can replay from any position, and many groups read the same records independently.

Clusters#

A StreamConfig says where the cluster is and how clients log in. Producers and consumers on the same cluster take the same config:

StreamConfig events = StreamConfig.builder("boot-abc123.c2.kafka-serverless.eu-west-2.amazonaws.com:9098")
        .auth(StreamAuth.iam())
        .build();

A supplied password, such as a read from a parameter store, takes effect without a restart:

StreamConfig.builder(bootstrapServers)
        .auth(StreamAuth.password(SaslMechanism.SCRAM_SHA_512, "orders-app",
                () -> secrets.require("AmazonMSK_orders-app", "password")))
        .build();

New connections log in with the current password, and open ones when the broker has them log in again (connections.max.reauth.ms, off by default). Without re-authentication, open connections stay logged in until they're replaced. The password never goes into sasl.jaas.config.

Cluster, producer and consumer properties that start with kafka. go to the Kafka clients without the prefix, so any client setting can be tuned: kafka.linger.ms, kafka.compression.type, kafka.max.poll.records. They override what Halyx and the provider set.

Sending#

StreamProducer orders = halyx.provider("aws").streamProducer(StreamProducerConfig.of(events, "orders"));

SendResult result = orders.send("customer-42", json);           // key and value
result.partition();
result.offset();

orders.send(OutgoingRecord.builder(json)
        .key("customer-42")
        .header("type", "created")
        .build());

orders.sendAll(List.of(OutgoingRecord.of("customer-42", a), OutgoingRecord.of("customer-7", b)));

• Keys choose the partition, so records with the same key are read in the order they were sent. Records without a key are spread across partitions.
• send waits until the broker has stored the record, and sendAll until it has stored all of them, with results in the same order.
• Values are bytes or text, and headers are string key-value pairs. Header names starting with halyx- are reserved.
• A producer is thread-safe. Keep one per topic; producers on the same cluster share one Kafka producer.
• partitionCount() asks the cluster for the topic's partitions, which also checks that it exists.

Sending without blocking#

async() gives a producer's asynchronous view: send and sendAll as CompletableFutures that complete from Kafka's send callbacks when the broker acknowledges, with claim checks stored through the blob store's asynchronous view. Kafka's client may still block a send briefly the first time it fetches a topic's metadata, or when its buffer is full (up to max.block.ms). A transactional producer runs each batch's transaction on a virtual thread.

Listening#

listen runs the consumer and hands each record to your handler:

StreamConsumer billing = halyx.provider("aws").streamConsumer(StreamConsumerConfig.builder(events, "orders", "billing")
        .startPosition(StartPosition.EARLIEST)
        .build());

StreamListener listener = billing.listen(record -> {
    bill(record.key().orElseThrow(), record.valueAsString());
}, ListenerOptions.DEFAULTS.withDeadLetter(deadLetters, 5));

// on shutdown: let running handlers finish, commit, and leave the group
billing.close();

• Partitions are handled in order, and in parallel. Each partition the consumer owns runs on its own virtual thread, one record at a time.
• Returning commits the record. Handled offsets are committed every second (withCommitInterval), and always before partitions are handed to another member or the listener stops. A lost commit only means records are read again.
• Throwing retries the same record with exponential backoff, 1 second doubling up to 5 minutes by default (withBackoff). The rest of its partition waits, so the order holds; other partitions carry on. The consumer keeps polling throughout, so a long retry never drops it from its group.
• Dead letters: with withDeadLetter(producer, maxAttempts), a record that fails maxAttempts times goes to that producer's topic, with halyx-dead-letter-topic, -partition, -offset and -error headers, and its partition moves on. Without one, a failing record is retried until it succeeds.
• Backpressure: each partition reads up to maxBufferedRecords (500) ahead of its handler, then pauses.
• listener.statistics() reports records received, handled, failed and dead-lettered, records in flight, read errors and the partitions this member owns.

Reading by hand#

poll returns the next records from the partitions this member owns, and commit records the position after everything poll has returned:

try (StreamConsumer audit = halyx.provider("aws").streamConsumer(StreamConsumerConfig.of(events, "orders", "audit"))) {
    while (running) {
        for (ReceivedRecord record : audit.poll(Duration.ofSeconds(1))) {
            archive(record);
        }
        audit.commit();
    }
}

Records polled but not committed are read again, by this member or whichever member owns the partition next. A consumer either polls by hand or runs a listener, not both, and poll and commit belong to one thread.

async() gives poll and commit as CompletableFutures, so the calling thread never waits. Kafka's consumer takes one call at a time, so a consumer's asynchronous calls run one after another, in the order they were made, each on a virtual thread: the view frees the caller's thread, not the consumer's.

Consumer groups#

• Every group reads every record. Consumers in the same group share the topic's partitions between them; start more members to read in parallel, up to one per partition.
• Where a new group starts is startPosition: LATEST (the default) reads only new records, EARLIEST everything the topic retains. A group with committed positions always resumes from them.
• Members leave the group on close(), and their partitions move to the others.
• A listener can be replaced. Closing a listener keeps the consumer in its group with its partitions; a new listen on the same consumer carries on from the first record the last one didn't handle.

Idempotence, transactions and large values#

StreamProducer payments = aws.streamProducer(StreamProducerConfig.builder(events, "payments")
        .transactional(true)            // sendAll is all or nothing
        .build());

StreamConsumer ledger = aws.streamConsumer(StreamConsumerConfig.builder(events, "payments", "ledger")
        .readCommitted(true)            // never see aborted records
        .build());

These are declared capabilities: asking for one the provider doesn't have fails rather than being ignored.

Event Hubs offers transactions only on its premium and dedicated tiers, as a preview, so Halyx doesn't declare them on Azure. Idempotent producers work on Event Hubs, Standard tier included, and are on by default there, as on the other clouds.

Values over a size threshold can go through a blob store with the same ClaimCheck as messaging, on both the producer and the consumer config. Without one, a value over the limit fails with InvalidArgumentException.

Provider notes#

• MSK: IAM clients use port 9098 (9198 from outside the VPC with public access). Tokens are signed for the connection's region; set aws.msk.region on the cluster if it differs. The identity needs kafka-cluster:Connect plus the topic and group actions it uses.
• Event Hubs: bootstrap <namespace>.servicebus.windows.net:9093. Tokens are for the namespace (azure.event-hubs.scope overrides it), and the identity needs the Azure Event Hubs Data Sender or Receiver role. Halyx applies Microsoft's recommended client settings, such as connections.max.idle.ms=180000, since Event Hubs closes idle connections after 240 seconds.
• Managed Kafka: the identity needs the Managed Kafka Client role. Service accounts are detected; for a gcloud user login, set gcp.kafka.principal to the account's email.

Native clients#

Producer<byte[], byte[]> producer = orders.unwrap(Producer.class);    // org.apache.kafka.clients.producer
Consumer<byte[], byte[]> consumer = billing.unwrap(Consumer.class);   // only from the thread that polls

The same on every provider. See Capabilities and native clients.

For Kafka's own clients, or any library that takes Kafka client properties, kafkaClientProperties gives the properties Halyx' producers and consumers use: bootstrap servers, TLS, and SASL with the cloud login (MSK IAM, Entra ID or Google IAM tokens, refreshed before they expire), with the provider's defaults and the cluster's kafka.* properties.

Map<String, Object> login = aws.kafkaClientProperties(events);   // events: the StreamConfig
try (Admin admin = Admin.create(login)) {
    admin.createTopics(List.of(new NewTopic("orders", 6, (short) 3))).all().get();
}

Some values are objects rather than strings (the login's token source), so pass the map to the client as it is. Spring for Apache Kafka uses the same properties.