Reading from Kafka¶

In addition to the HTTP API, the application can consume FHIR resources/bundles directly from one or more Kafka topics, pseudonymize them, and publish the result to an output topic.

This requires setting Kafka__Enabled=true, Kafka__Client__BootstrapServers, and at least one topic in Kafka__Topics. See the options below.

Output topics and keys¶

By default, the output message's key is not the original message's key but its crypto-hash - the lowercase hex HMAC-SHA256, keyed with Kafka__CryptoHashMessageKeys__Key - since keys are often identifiers, e.g. a patient's id. The same key is always hashed to the same value (as long as Kafka__CryptoHashMessageKeys__Key doesn't change), so messages that shared a key still share one after pseudonymization, and with it their output partition. If Kafka__CryptoHashMessageKeys__Key is the same as the anonymization config's cryptoHashKey (with the default hmacSha256 algorithm and no key derivation), a message key is hashed to the same value as a cryptoHash rule turns the same text into. Set Kafka__CryptoHashMessageKeys__Enabled=false to keep the original keys instead.

The output topic for a given input topic is derived by applying a regular expression match-and-replace (Kafka__OutputTopicPattern / Kafka__OutputTopicReplacement) to the input topic's name. By default, this just prepends pseudonymized. to every topic, e.g. fhir.test becomes pseudonymized.fhir.test. To instead insert it after a common prefix, e.g. turning fhir.test into fhir.pseudonymized.test, set Kafka__OutputTopicPattern=^fhir\. and Kafka__OutputTopicReplacement=fhir.pseudonymized..

Concurrency¶

Messages are anonymized concurrently by Kafka__WorkerCount consumers (defaulting to the number of CPU cores), all members of the same consumer group.

Each consumer prefetches up to Kafka__Consumer__QueuedMaxMessagesKbytes worth of messages (16 MiB unless set, instead of librdkafka's default of 64 MiB, since there is one consumer per worker). A consumer that is busy simply stops fetching more, without holding up the others.

Failures and dead-lettering¶

Transient failures calling the configured pseudonymization service (gPAS, Vfps, entici, MII) are retried with backoff at the HTTP/gRPC client level and, if they persist, the message is retried indefinitely (with exponential backoff of up to one minute) instead of being dead-lettered, since the message itself isn't at fault. Meanwhile, its partition is paused, while the consumer carries on with its other partitions.

If processing a message fails for any other reason (e.g. malformed input, the pseudonymization service rejecting a value - see the 422 status - or an unrecoverable error), or its pseudonymized version can't be delivered to the output topic (or its provenance to Kafka__ProvenanceTopic, see Provenance), the original, unmodified message is published to a dead letter topic named error.<input-topic>.<group-id> (mirroring Spring Kafka's default dead letter topic naming), with headers describing the failure (x-error-type, x-error-message, x-source-topic, x-source-partition, x-source-offset).

Delivery guarantees¶

A message's offset is only committed once the broker has acknowledged the message it was turned into - on the output or the dead letter topic - and those of all earlier messages of the same partition, too. This way, a crash or restart never skips a message, though it may process some of them twice: after a crash, the messages processed since the last commit (every Kafka__Consumer__AutoCommitIntervalMs, 5 seconds by default) are processed again.

When a rebalance moves partitions to another replica instead, e.g. when scaling up or down, their queued messages are dropped, but the ones already being processed are given up to 10 seconds to finish, and their offsets committed, before the partitions are handed over, so this doesn't produce duplicates. If even publishing a message to the dead letter topic fails, the application stops instead of moving past it, so that it is reprocessed after a restart.

Validation¶

Messages are parsed and serialized using the System.Text.Json-based FHIR serializer. It validates resources while parsing them, so a message that isn't valid FHIR (e.g. a resource with an id containing characters FHIR doesn't allow, or missing required elements) is published to the dead letter topic.

Options¶