whitewater-java

Clients

WhitewaterClient client = WhitewaterClient.builder()
        .apiKey(System.getenv("WHITEWATER_API_KEY"))
        .endpoints(List.of(
                URI.create("https://node1:7070"),
                URI.create("https://node2:7070"),
                URI.create("https://node3:7070")))
        .build();

var accounts = client.createDomain("accounts");
var spec = StateStoreSpec.named("employees")
        .addUniqueKey("passport").addNonUniqueKey("department");          // up to 10 named keys
var store = accounts.defineStateStore(spec, null);                    // durable store on the Nodes

// named-key postings: pick whichever call-site style reads cleanest —
store.put("emp-1", payload, Map.of("passport", "P123", "department", "eng")); // raw map
store.put("emp-1", payload, PassportKeySet.with("P123", "eng"));              // typed key set
store.put("emp-1", payload, PassportKeySet.of(k -> k.passport("P123").department("eng")));
store.put("emp-1", payload, "P123");                                          // single named key
var row = store.get("emp-1");                          // StoreRow | null + freshness
var rows = store.getBy("passport", "P123");            // named key — a list
store.iterate(r -> export(r));

var demo = client.createDomain("demo");
Feed<String, Order> orders = demo.feed("orders", Order.class);
orders.writer("ingest").append("order-1", new Order("order-1", 42));
try (var reader = orders.reader("audit", 64)) {
    Order o = reader.fetch().records().get(0).value();   // decoded
}
StateStore<String, Order> latest = demo.stateStore("orders_latest", Order.class);
latest.put("order-1", order);                            // same codec seam

Domains — the strongly-typed front door

import static dev.whitewater.model.ReaderStart.*;   // Beginning, End

client.createDomain("shop");                        // CREATE DOMAIN shop;
Domain shop = client.getDomain("shop");             // verifies + materializes now
shop.createFeed("orders");                          // CREATE FEED shop.orders;
shop.feedNames();                                 // [shop.orders, shop.returns]
shop.getFeeds("order");                           // prefix-filtered Feed handles
shop.readers(); shop.writers(); shop.subscriptions();

// Everything a service needs hangs off the Domain:
var writer = shop.createWriter("orders")
        .writerId("ingest")
        .open();
var reader = shop.createReader("orders")
        .readerId("fulfillment")                    // durable consume identity
        .readerStartPoint(Beginning)                // first-time start position
        .maxPageSize(250)
        .open();
try (var pump = reader.listen(m -> handle(m))) {    // SDK-owned thread — returns now
    // … the service's other work runs untouched; pump.failure() surfaces
    //    a handler error; pump.close() stops + closes, page unacked → replays
}
// or stay manual: fetch() → handle records → ack()

Record schemas

shop.setSchema(
        FeedSchema.value(SchemaDecl.of("json", "order-v1")));
shop.setFeedSchema("orders",
        FeedSchema.key(SchemaDecl.of("text")));

shop.feedSchema("orders");          // effective: key=text, value=json/order-v1
shop.declaredFeedSchema("orders");  // the Feed's own sides only
shop.schema();                      // the Domain default
shop.clearFeedSchema("orders");     // reverts to inheritance

Permanent consumers — the Kafka loop, minus the plumbing

var consumer = orders.consumer("enricher")
        .onRecord(msg -> process(msg))              // one record at a time
        .onError(err -> log.warn("retrying", err)); // visibility only
Runtime.getRuntime().addShutdownHook(new Thread(consumer::close));
consumer.run();          // blocks until close() — that's the whole loop

var user = users.find(email)                         // Optional<UserProfile>
        .or(() -> users.findBy("user_id", id));      // named-key fallback
counters.put(bucket, counters.getOrDefault(bucket, 0L) + 1);

Pipes, indexes, and operations

// Pipe declaration inside the Domain: names resolve in-domain (orders.relay
// is ops.orders.relay when declared on ops), and cross-Domain endpoints
// ride on ScopedRef — a Feed/StateStore/SubscriptionRef handle, or
// Scoped.absolute. The server owns panes, leases and the effect journal.
var ops = client.domain("ops");
var crm = client.domain("crm");
ops.defineForwardPipe("orders.relay",
        Scoped.relative("orders-sub"), ops.feed("orders.relayed"));
ops.defineEnrichPipe("orders.enrich",
        Scoped.relative("orders-sub"), ops.feed("orders.enriched"),
        crm.stateStore("customers"),                       // cross-Domain lookup
        PipeOperation.enrich(Extractor.recordKey(),
            EnrichFreshness.LATEST, null, EnrichMiss.QUARANTINE),
        ops.feed("orders.enrich-dlq"));
ops.defineWindowPipe("orders.hourly",
        Scoped.relative("orders-sub"), Scoped.relative("stats.orders-hourly"),
        PipeOperation.window(Time.h(1),   // 1h panes
            Extractor.recordKey(), WindowAggregation.count(),
            WindowTime.EVENT,
            null,                 // maxLate
            Time.m(5),            // watermark: 5m horizon
            Time.s(30),           // grace: 30s late revisions
            Time.h(2),            // expire
            null, null),          // keyLimit, byteLimit
        ops.feed("orders.hourly-emitted"));

// Pipe lifecycle: describe, pause, resume, drop.
var pipe = ops.pipe("orders.relay");
PipeInfo info = pipe.describe();   // stage, operation, outputs, status
pipe.pause();
pipe.resume();
List<PipeInfo> all = ops.pipeInfos();

// Index queries: exact, prefix, or range predicates over decoded rows.
IndexPage page = crm.queryIndex(
        "byregion",
        IndexPredicate.range(
                List.of("emea"),
                IndexPredicate.IndexBound.inclusive(0),
                IndexPredicate.IndexBound.exclusive(99)),
        50,        // limit
        true,      // strict: wait for the committed target rather than stale
        null);     // wait_ms
for (var row : page.rows()) { /* key + payload bytes, applied cursor */ }

// Demand observability: typed rows + raw domain rollups.
DemandReport feeds = client.demandFeeds();
for (DemandRow row : feeds.rows()) {
    long appended = row.counter("appended_records");
}
client.demandReaders(); client.demandSubscriptions();
client.demandIndexes();  client.demandPipes();
List<HotKey> hot = client.hotKeys();   // dominant-key evidence per feed

API surface

Build

mvn verify          # compile + unit tests (+ conformance, if configured)

docker run --rm -v "${PWD}:/work" -w /work maven:3.9-eclipse-temurin-21 mvn verify

Unit-test doubles

try (var riverbed = FakeRiverbed.start();
     var client = riverbed.client()) { ... }

Conformance

$env:WHITEWATER_ENDPOINTS = "http://localhost:7071,http://localhost:7072,http://localhost:7073"
$env:WHITEWATER_API_KEY   = "whitewater-local-development-admin-key"
$env:WHITEWATER_SPEC_DIR  = "C:\Code\Stackapps\whitewater\sdk\spec\conformance"
mvn verify