shivamka1

# Producer (web request handler)
function request_purchase(productId):
  enqueue(topic="purchases", key=productId, value="BUY")

# Consumer (single-threaded per key-partition)
function purchase_worker():
  loop:
    messages = poll(topic="purchases", timeout_ms=200)
    for msg in messages:           # all messages for same key arrive in order
      productId = msg.key

shivamka1

function acquire_distributed_lock(key, ttl_ms) -> (ok, token):
  # e.g. SET key token NX PX ttl_ms in Redis
  ...

function release_distributed_lock(key, token):
  # delete key only if stored token matches
  ...

function purchase(productId):
  (ok, token) = acquire_distributed_lock("lock:product:" + productId, ttl_ms=5000)

shivamka1

locks = Map<productId, LockObject>()

function get_lock(productId):
  if productId not in locks:
    locks[productId] = new LockObject()
  return locks[productId]

function purchase(productId):
  lock = get_lock(productId)

shivamka1

function purchase(productId):
  BEGIN TRANSACTION (ISOLATION = SERIALIZABLE)

  stock = query("SELECT stock FROM products WHERE id = ?", productId)

  if stock <= 0:
    ROLLBACK
    return "Out of stock"

  exec("UPDATE products SET stock = stock - 1 WHERE id = ?", productId)

shivamka1

#!/usr/bin/env python3
"""
Bloom filter drift plots: false-positive rate vs n, bits-per-key vs n,
optimal k vs n, and a combined figure.

Usage (defaults to n0=1_000_000, p0=0.01):
    python bloom_drift_plots.py
    python bloom_drift_plots.py --n0 2000000 --p0 0.005

Outputs (.png and .svg):

shivamka1

% /opt/zookeeper/current/bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 3] ls /flightServers/flightServer
[978ead9a-64d3-4080-9163-235cf99fe1b9, a89b2126-27cb-43c2-a687-2d8d57d74077, d87c5515-4d26-46fa-bc02-06edcc04f86a]
[zk: localhost:2181(CONNECTED) 4] get /flightServers/flightServer/978ead9a-64d3-4080-9163-235cf99fe1b9
{
  "name": "flightServer",
  "id": "978ead9a-64d3-4080-9163-235cf99fe1b9",
  "address": "127.0.0.1",
  "port": 8877,

shivamka1

serviceDiscovery.registerService(instance)

shivamka1

val instance =
    ServiceInstance.builder()
      .address(InetAddress.getLocalHost.getHostAddress)
      .port(8855)
      .name("flightServer")
      .payload(ServiceDetail(partitionId)).build()
      

shivamka1

val ServiceDiscoveryAtomicPath = "/flightServers"

val serviceDiscovery =
    ServiceDiscoveryBuilder
      .builder(classOf[ServiceDetail])
      .client(zkClient)
      .serializer(new JsonInstanceSerializer[ServiceDetail](classOf[ServiceDetail]))
      .basePath(ServiceDiscoveryAtomicPath)
      .build()

shivamka1

class ServiceDetail() {
  private var partitionId: Int = 0

  def this(partitionId: Int) {
    this()
    this.partitionId = partitionId
  }

  def getPartitionId: Int = partitionId