Skip to main content

Keys

A record's partition key picks its shard and its order; four places can name it, and a record nobody keyed gets a process-unique one.

The service hashes the key to pick the shard, so every record of one key lands on one shard, and a reader sees them in the order they were written. A key is 1 to 256 characters; the service refuses a record with any other with KinesisError::Publish. One key is one shard's write throughput, which the service limits per shard, so a key that every record shares orders them all at the cost of that throughput.

The publisher takes the first key it finds, in this order:

  1. the partition_key step of the publish;
  2. the partition-key header, PARTITION_KEY_HEADER, that the message carries;
  3. the key the mount site named on the policy, with KinesisPublishSettings;
  4. a key unique to the process, which spreads the records over the shards.

The header is the portable spelling: a body that names the key that way stays free of this crate, and a message that crossed from another broker of the framework carries its key in it. A message type declares the header in its typed header contract, #[outgoing(headers = ..)], with a field renamed to partition-key; a HeaderMap with that entry works the same. The key travels in the record's own partition-key field, and the header is left out of the envelope. On delivery the subscription reports the key under the same header.

Example​

Every shipment of one customer carries the customer as its key, so the warehouse reads them in order.

examples/kinesis_publishing_keys.rs
/// The header a shipment carries: `partition-key`, which becomes the record's partition key.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct ShipmentKey {
    #[serde(rename = "partition-key")]
    customer: String,
}

/// Every shipment of one customer lands on one shard, so its consumer sees them in order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments", headers = ShipmentKey)]
struct Shipment {
    order: u64,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

/// The body names no broker: the plain bound moves between brokers untouched.
#[subscriber(KinesisStream::new("orders"))]
async fn ship(order: &Order, Out(shipments): Out<impl Publisher, Shipments>) -> HandlerOutcome {
    let key = ShipmentKey {
        customer: order.customer.clone(),
    };
    let shipment = Shipment { order: order.id };
    if shipments
        .message(&shipment)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}