data pipeline refactor and fix

flink/src/main/java/com/dexorder/flink/TradingFlinkApp.java

@@ -2,7 +2,8 @@ package com.dexorder.flink;
 
 import com.dexorder.flink.config.AppConfig;
 import com.dexorder.flink.iceberg.SchemaInitializer;
-import com.dexorder.flink.ingestor.IngestorWorkQueue;
+import com.dexorder.flink.ingestor.IngestorBroker;
+import com.dexorder.flink.ingestor.RealtimeSubscriptionManager;
 import com.dexorder.flink.kafka.TopicManager;
 import com.dexorder.flink.publisher.HistoryNotificationForwarder;
 import com.dexorder.flink.publisher.HistoryNotificationFunction;
@@ -10,6 +11,11 @@ import com.dexorder.flink.publisher.OHLCBatchWrapper;
 import com.dexorder.flink.publisher.OHLCBatchDeserializer;
 import com.dexorder.flink.publisher.MarketWrapper;
 import com.dexorder.flink.publisher.MarketDeserializer;
+import com.dexorder.flink.publisher.RealtimeBar;
+import com.dexorder.flink.publisher.RealtimeBarFunction;
+import com.dexorder.flink.publisher.RealtimeBarPublisher;
+import com.dexorder.flink.publisher.TickWrapper;
+import com.dexorder.flink.publisher.TickDeserializer;
 import com.dexorder.flink.sink.HistoricalBatchWriter;
 import com.dexorder.flink.sink.SymbolMetadataWriter;
 import com.dexorder.flink.zmq.ZmqChannelManager;
@@ -83,11 +89,16 @@ public class TradingFlinkApp {
                     catalogProps
             );
 
+            String warehouse = config.getString("iceberg_warehouse", "s3://warehouse/");
+            String warehouseBucket = warehouse.replaceFirst("^s3://", "").split("/")[0];
+
             org.apache.iceberg.catalog.Catalog catalog = catalogLoader.loadCatalog();
             try {
                 SchemaInitializer schemaInitializer = new SchemaInitializer(
                         catalog,
-                        config.getIcebergNamespace()
+                        config.getIcebergNamespace(),
+                        config.getString("s3_endpoint", "http://minio:9000"),
+                        warehouseBucket
                 );
                 schemaInitializer.initializeSchemas();
             } finally {
@@ -107,20 +118,28 @@ public class TradingFlinkApp {
             zmqManager.initializeChannels();
             LOG.info("ZeroMQ channels initialized");
 
-            // Initialize history notification forwarder (runs in job manager)
-            // Binds PULL socket to receive notifications from task managers, forwards to MARKET_DATA_PUB
+            // Initialize ingestor broker — manages ROUTER/DEALER work queue for all ingestors
+            IngestorBroker broker = new IngestorBroker(zmqManager);
+            broker.start();
+            LOG.info("IngestorBroker started");
+
+            // Initialize realtime subscription manager — owns MARKET_DATA_PUB socket exclusively,
+            // detects XPUB subscription events, and calls broker for realtime job lifecycle.
+            // Other components publish via subscriptionManager.enqueuePublish() (thread-safe).
+            RealtimeSubscriptionManager subscriptionManager = new RealtimeSubscriptionManager(zmqManager, broker);
+            subscriptionManager.start();
+            LOG.info("RealtimeSubscriptionManager started");
+
+            // Initialize history notification forwarder (runs in job manager).
+            // Binds PULL socket to receive notifications from task managers, enqueues them for
+            // publication via RealtimeSubscriptionManager (sole owner of MARKET_DATA_PUB).
             HistoryNotificationForwarder notificationForwarder = new HistoryNotificationForwarder(
                     config.getNotificationPullPort(),
-                    zmqManager.getSocket(ZmqChannelManager.Channel.MARKET_DATA_PUB)
+                    subscriptionManager::enqueuePublish
             );
             notificationForwarder.start();
             LOG.info("History notification forwarder started on port {}", config.getNotificationPullPort());
 
-            // Initialize ingestor work queue
-            IngestorWorkQueue workQueue = new IngestorWorkQueue(zmqManager);
-            workQueue.start();
-            LOG.info("Ingestor work queue started");
-
             // Set up Flink streaming environment
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
@@ -224,8 +243,37 @@ public class TradingFlinkApp {
 
             LOG.info("Symbol metadata pipeline configured: SymbolMetadataWriter -> Iceberg -> METADATA_UPDATE notification");
 
+            // Realtime tick pipeline: Kafka market-tick → OHLC bars → ZMQ notify → clients
+            KafkaSource<TickWrapper> tickSource = KafkaSource.<TickWrapper>builder()
+                    .setBootstrapServers(config.getKafkaBootstrapServers())
+                    .setTopics(config.getKafkaTickTopic())
+                    .setGroupId("flink-tick-consumer")
+                    .setStartingOffsets(OffsetsInitializer.latest())
+                    .setValueOnlyDeserializer(new TickDeserializer())
+                    .build();
+
+            DataStream<TickWrapper> tickStream = env
+                    .fromSource(tickSource, WatermarkStrategy.noWatermarks(), "Tick Kafka Source")
+                    .filter(t -> t != null)
+                    .setParallelism(1);
+
+            // Aggregate ticks into OHLC bars for each configured period.
+            // keyBy ticker so all ticks for a ticker land on the same slot and accumulate together.
+            int[] periods = config.getRealtimePeriods();
+
+            DataStream<RealtimeBar> barStream = tickStream
+                    .keyBy(TickWrapper::getTicker)
+                    .flatMap(new RealtimeBarFunction(periods))
+                    .setParallelism(1);
+
+            barStream.addSink(new RealtimeBarPublisher(notificationEndpoint))
+                    .setParallelism(1)
+                    .name("RealtimeBarPublisher");
+
+            LOG.info("Realtime tick pipeline configured: market-tick → OHLC bars → clients (periods={})",
+                    java.util.Arrays.toString(periods));
+
             // TODO: Set up CEP patterns and triggers
-            // TODO: Set up realtime tick processing
 
             LOG.info("Flink job configured, starting execution");
 
@@ -233,15 +281,10 @@ public class TradingFlinkApp {
             Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                 LOG.info("Shutting down Trading Flink Application");
                 try {
-                    // Stop work queue
-                    workQueue.stop();
-
-                    // Stop notification forwarder
                     notificationForwarder.close();
-
-                    // Close ZMQ channels
+                    subscriptionManager.stop();
+                    broker.stop();
                     zmqManager.close();
-
                     LOG.info("Shutdown complete");
                 } catch (Exception e) {
                     LOG.error("Error during shutdown", e);

flink/src/main/java/com/dexorder/flink/config/AppConfig.java

@@ -91,14 +91,20 @@ public class AppConfig {
     }
 
     // ZMQ port getters
-    public int getIngestorWorkQueuePort() {
-        return getInt("zmq_ingestor_work_queue_port", 5555);
-    }
-
     public int getMarketDataPubPort() {
         return getInt("zmq_market_data_pub_port", 5558);
     }
 
+    /** Port where Flink's IngestorBroker binds a PULL socket to receive requests from relay PUSH */
+    public int getFlinkRequestPullPort() {
+        return getInt("zmq_flink_request_pull_port", 5566);
+    }
+
+    /** Port where Flink's IngestorBroker binds a ROUTER for ingestor DEALER connections */
+    public int getIngestorBrokerPort() {
+        return getInt("zmq_ingestor_broker_port", 5567);
+    }
+
     public String getBindAddress() {
         return getString("zmq_bind_address", "tcp://*");
     }
@@ -112,6 +118,20 @@ public class AppConfig {
         return getString("kafka_tick_topic", "market-tick");
     }
 
+    /**
+     * Comma-separated OHLC period lengths in seconds for realtime bar computation.
+     * Default covers common chart periods: 1m, 5m, 15m, 1h, 4h, 1d.
+     */
+    public int[] getRealtimePeriods() {
+        String raw = getString("realtime_periods", "60,300,900,3600,14400,86400");
+        String[] parts = raw.split(",");
+        int[] periods = new int[parts.length];
+        for (int i = 0; i < parts.length; i++) {
+            periods[i] = Integer.parseInt(parts[i].trim());
+        }
+        return periods;
+    }
+
     public String getKafkaOhlcTopic() {
         return getString("kafka_ohlc_topic", "market-ohlc");
     }

flink/src/main/java/com/dexorder/flink/iceberg/SchemaInitializer.java

@@ -9,8 +9,16 @@ import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.types.Types;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3Configuration;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.HeadBucketRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchBucketException;
 
 import java.io.IOException;
+import java.net.URI;
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.apache.iceberg.types.Types.NestedField.required;
@@ -26,10 +34,14 @@ public class SchemaInitializer {
 
     private final Catalog catalog;
     private final String namespace;
+    private final String s3Endpoint;
+    private final String warehouseBucket;
 
-    public SchemaInitializer(Catalog catalog, String namespace) {
+    public SchemaInitializer(Catalog catalog, String namespace, String s3Endpoint, String warehouseBucket) {
         this.catalog = catalog;
         this.namespace = namespace;
+        this.s3Endpoint = s3Endpoint;
+        this.warehouseBucket = warehouseBucket;
     }
 
     /**
@@ -40,6 +52,9 @@ public class SchemaInitializer {
     public void initializeSchemas() throws IOException {
         LOG.info("Initializing Iceberg schemas in namespace: {}", namespace);
 
+        // Ensure S3 bucket exists before attempting to create tables
+        ensureS3BucketExists();
+
         // Ensure namespace exists
         ensureNamespaceExists();
 
@@ -52,6 +67,36 @@ public class SchemaInitializer {
         LOG.info("Schema initialization completed successfully");
     }
 
+    /**
+     * Ensure the S3 warehouse bucket exists, creating it if necessary.
+     * Runs before any table creation so a fresh MinIO deployment doesn't crash Flink.
+     */
+    private void ensureS3BucketExists() {
+        if (s3Endpoint == null || warehouseBucket == null || warehouseBucket.isEmpty()) {
+            LOG.warn("S3 endpoint or warehouse bucket not configured, skipping bucket check");
+            return;
+        }
+        LOG.info("Ensuring S3 bucket '{}' exists at {}", warehouseBucket, s3Endpoint);
+        try (S3Client s3 = S3Client.builder()
+                .endpointOverride(URI.create(s3Endpoint))
+                .region(Region.of("us-east-1"))
+                .serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build())
+                .credentialsProvider(DefaultCredentialsProvider.create())
+                .build()) {
+            try {
+                s3.headBucket(HeadBucketRequest.builder().bucket(warehouseBucket).build());
+                LOG.info("S3 bucket '{}' already exists", warehouseBucket);
+            } catch (NoSuchBucketException e) {
+                LOG.warn("S3 bucket '{}' not found — creating it now", warehouseBucket);
+                s3.createBucket(CreateBucketRequest.builder().bucket(warehouseBucket).build());
+                LOG.info("Created S3 bucket '{}'", warehouseBucket);
+            }
+        } catch (Exception e) {
+            LOG.error("Failed to ensure S3 bucket '{}' exists at {}", warehouseBucket, s3Endpoint, e);
+            throw new RuntimeException("S3 bucket initialization failed for: " + warehouseBucket, e);
+        }
+    }
+
     /**
      * Ensure the namespace exists in the catalog.
      */

flink/src/main/java/com/dexorder/flink/ingestor/IngestorBroker.java

@@ -0,0 +1,503 @@
+package com.dexorder.flink.ingestor;
+
+import com.dexorder.flink.zmq.ZmqChannelManager;
+import com.dexorder.proto.DataRequest;
+import com.dexorder.proto.RealtimeParams;
+import com.dexorder.proto.SubmitHistoricalRequest;
+import com.dexorder.proto.WorkComplete;
+import com.dexorder.proto.WorkHeartbeat;
+import com.dexorder.proto.WorkReject;
+import com.dexorder.proto.WorkStop;
+import com.dexorder.proto.WorkerReady;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.zeromq.ZMQ;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * LRU-style work broker for ingestors.
+ *
+ * Ingestors connect via DEALER to the ROUTER socket on port 5567. They register with READY,
+ * are dispatched WORK messages, and respond with COMPLETE (historical) or HEARTBEAT (realtime).
+ * If a heartbeat times out the job is re-queued and dispatched to another available worker.
+ *
+ * Also receives SubmitHistoricalRequest messages forwarded by the relay on the PULL socket (5566).
+ *
+ * Message type IDs (ZMQ framing, not Kafka):
+ *   0x10  SubmitHistoricalRequest (relay → Flink via PULL, same as client wire type)
+ *   0x20  WorkerReady             (ingestor → Flink)
+ *   0x21  WorkComplete            (ingestor → Flink)
+ *   0x22  WorkHeartbeat           (ingestor → Flink)
+ *   0x23  WorkReject              (ingestor → Flink)
+ *   0x01  DataRequest/WorkAssign  (Flink → ingestor via ROUTER)
+ *   0x25  WorkStop                (Flink → ingestor via ROUTER)
+ */
+public class IngestorBroker implements AutoCloseable {
+    private static final Logger LOG = LoggerFactory.getLogger(IngestorBroker.class);
+
+    private static final byte PROTOCOL_VERSION = 0x01;
+    private static final byte MSG_TYPE_SUBMIT_REQUEST   = 0x10;
+    private static final byte MSG_TYPE_WORKER_READY     = 0x20;
+    private static final byte MSG_TYPE_WORK_COMPLETE    = 0x21;
+    private static final byte MSG_TYPE_WORK_HEARTBEAT   = 0x22;
+    private static final byte MSG_TYPE_WORK_REJECT      = 0x23;
+    private static final byte MSG_TYPE_WORK_ASSIGN      = 0x01;  // DataRequest type on wire
+    private static final byte MSG_TYPE_WORK_STOP        = 0x25;
+
+    /** Re-queue realtime job if no heartbeat received within this window (ms) */
+    private static final long HEARTBEAT_TIMEOUT_MS = 25_000;
+    /** Re-queue historical job if not completed within this window (ms) */
+    private static final long HISTORICAL_TIMEOUT_MS = 60_000;
+
+    private final ZmqChannelManager zmqManager;
+    private volatile boolean running;
+    private Thread brokerThread;
+
+    // ── Worker tracking ──────────────────────────────────────────────────────
+
+    /** Workers ready to accept a job, in LRU order (head = least recently used) */
+    private final Deque<WorkerInfo> freeWorkers = new ArrayDeque<>();
+
+    /** Jobs waiting for a compatible free worker */
+    private final Queue<DataRequest> pendingJobs = new ArrayDeque<>();
+
+    /** Jobs currently executing on a worker */
+    private final Map<String, ActiveJob> activeJobs = new ConcurrentHashMap<>();
+
+    /** Worker identity → supported exchanges (set once on READY) */
+    private final Map<String, WorkerInfo> knownWorkers = new ConcurrentHashMap<>();
+
+    // ── Thread-safe inbound queue from RealtimeSubscriptionManager ───────────
+
+    private final Queue<DataRequest> externalSubmissions = new ConcurrentLinkedQueue<>();
+
+    public IngestorBroker(ZmqChannelManager zmqManager) {
+        this.zmqManager = zmqManager;
+    }
+
+    public void start() {
+        if (running) {
+            LOG.warn("IngestorBroker already running");
+            return;
+        }
+        running = true;
+        brokerThread = new Thread(this::brokerLoop, "IngestorBroker-Thread");
+        brokerThread.setDaemon(false);
+        brokerThread.start();
+        LOG.info("IngestorBroker started");
+    }
+
+    public void stop() {
+        running = false;
+        if (brokerThread != null) {
+            brokerThread.interrupt();
+            try {
+                brokerThread.join(5000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        LOG.info("IngestorBroker stopped");
+    }
+
+    /**
+     * Submit a realtime data request from outside the broker thread (thread-safe).
+     * Called by RealtimeSubscriptionManager when subscription ref count goes 0→1.
+     */
+    public void submitRealtimeRequest(String ticker) {
+        String jobId = UUID.randomUUID().toString();
+        DataRequest request = DataRequest.newBuilder()
+                .setRequestId(jobId)
+                .setJobId(jobId)
+                .setType(DataRequest.RequestType.REALTIME_TICKS)
+                .setTicker(ticker)
+                .setRealtime(RealtimeParams.newBuilder()
+                        .setIncludeTicks(true)
+                        .setIncludeOhlc(false)
+                        .build())
+                .build();
+        externalSubmissions.add(request);
+        LOG.info("Enqueued realtime request: ticker={}, jobId={}", ticker, jobId);
+    }
+
+    /**
+     * Stop all realtime jobs for a ticker (called when last subscriber leaves).
+     * Thread-safe — posts a stop marker via externalSubmissions is complex; instead we
+     * directly find and stop active jobs. Protected by ConcurrentHashMap.
+     */
+    public void stopRealtimeJobsForTicker(String ticker) {
+        List<String> toStop = new ArrayList<>();
+        for (Map.Entry<String, ActiveJob> entry : activeJobs.entrySet()) {
+            if (entry.getValue().ticker.equals(ticker) &&
+                entry.getValue().type == DataRequest.RequestType.REALTIME_TICKS) {
+                toStop.add(entry.getKey());
+            }
+        }
+        for (String jobId : toStop) {
+            ActiveJob job = activeJobs.remove(jobId);
+            if (job != null) {
+                sendStop(job.workerIdentity, jobId);
+                LOG.info("Sent STOP to ingestor: ticker={}, jobId={}", ticker, jobId);
+            }
+        }
+    }
+
+    // ── Broker loop ──────────────────────────────────────────────────────────
+
+    private void brokerLoop() {
+        ZMQ.Socket pullSocket   = zmqManager.getSocket(ZmqChannelManager.Channel.CLIENT_REQUEST);
+        ZMQ.Socket routerSocket = zmqManager.getSocket(ZmqChannelManager.Channel.INGESTOR_BROKER);
+
+        ZMQ.Poller poller = zmqManager.createPoller(2);
+        poller.register(pullSocket, ZMQ.Poller.POLLIN);
+        poller.register(routerSocket, ZMQ.Poller.POLLIN);
+
+        LOG.info("IngestorBroker loop running");
+
+        while (running) {
+            try {
+                // Drain external submissions (realtime requests from subscription manager)
+                DataRequest ext;
+                while ((ext = externalSubmissions.poll()) != null) {
+                    enqueueJob(ext);
+                }
+
+                // Poll sockets (100ms timeout)
+                poller.poll(100);
+
+                if (poller.pollin(0)) {
+                    handleClientRequest(pullSocket);
+                }
+
+                if (poller.pollin(1)) {
+                    handleWorkerMessage(routerSocket);
+                }
+
+                // Check for heartbeat / completion timeouts
+                checkTimeouts();
+
+            } catch (Exception e) {
+                if (running) {
+                    LOG.error("Error in broker loop", e);
+                }
+            }
+        }
+
+        LOG.info("IngestorBroker loop exited");
+    }
+
+    /** Receive a SubmitHistoricalRequest forwarded by relay and enqueue it. */
+    private void handleClientRequest(ZMQ.Socket pullSocket) {
+        byte[] versionFrame = pullSocket.recv(ZMQ.DONTWAIT);
+        if (versionFrame == null) return;
+        if (!pullSocket.hasReceiveMore()) return;
+        byte[] messageFrame = pullSocket.recv(0);
+        if (messageFrame == null || messageFrame.length < 2) return;
+
+        if (versionFrame.length != 1 || versionFrame[0] != PROTOCOL_VERSION) {
+            LOG.warn("Bad protocol version on PULL socket");
+            return;
+        }
+
+        byte msgType = messageFrame[0];
+        byte[] payload = Arrays.copyOfRange(messageFrame, 1, messageFrame.length);
+
+        if (msgType != MSG_TYPE_SUBMIT_REQUEST) {
+            LOG.warn("Unexpected message type on PULL socket: 0x{}", Integer.toHexString(msgType & 0xFF));
+            return;
+        }
+
+        try {
+            SubmitHistoricalRequest req = SubmitHistoricalRequest.parseFrom(payload);
+            String jobId = UUID.randomUUID().toString();
+            DataRequest dataRequest = DataRequest.newBuilder()
+                    .setRequestId(req.getRequestId())
+                    .setJobId(jobId)
+                    .setType(DataRequest.RequestType.HISTORICAL_OHLC)
+                    .setTicker(req.getTicker())
+                    .setHistorical(com.dexorder.proto.HistoricalParams.newBuilder()
+                            .setStartTime(req.getStartTime())
+                            .setEndTime(req.getEndTime())
+                            .setPeriodSeconds(req.getPeriodSeconds())
+                            .build())
+                    .setClientId(req.hasClientId() ? req.getClientId() : "")
+                    .build();
+            enqueueJob(dataRequest);
+            LOG.info("Received historical request from relay: request_id={}, ticker={}", req.getRequestId(), req.getTicker());
+        } catch (Exception e) {
+            LOG.error("Failed to parse SubmitHistoricalRequest from relay", e);
+        }
+    }
+
+    /** Receive and dispatch a message from an ingestor DEALER. */
+    private void handleWorkerMessage(ZMQ.Socket routerSocket) {
+        // ROUTER frame layout: [identity][empty][version][type+payload]
+        byte[] identity = routerSocket.recv(ZMQ.DONTWAIT);
+        if (identity == null) return;
+        if (!routerSocket.hasReceiveMore()) return;
+        routerSocket.recv(0); // empty delimiter
+        if (!routerSocket.hasReceiveMore()) return;
+        byte[] versionFrame = routerSocket.recv(0);
+        if (!routerSocket.hasReceiveMore()) return;
+        byte[] messageFrame = routerSocket.recv(0);
+
+        if (versionFrame == null || versionFrame.length != 1 || versionFrame[0] != PROTOCOL_VERSION) {
+            LOG.warn("Bad protocol version from ingestor");
+            return;
+        }
+        if (messageFrame == null || messageFrame.length < 1) return;
+
+        byte msgType = messageFrame[0];
+        byte[] payload = Arrays.copyOfRange(messageFrame, 1, messageFrame.length);
+        String identityKey = bytesToHex(identity);
+
+        try {
+            switch (msgType & 0xFF) {
+                case 0x20: handleWorkerReady(identity, identityKey, payload); break;
+                case 0x21: handleWorkComplete(identityKey, payload); break;
+                case 0x22: handleWorkHeartbeat(identityKey, payload); break;
+                case 0x23: handleWorkReject(identityKey, payload); break;
+                default:
+                    LOG.warn("Unknown message type from ingestor: 0x{}", Integer.toHexString(msgType & 0xFF));
+            }
+        } catch (Exception e) {
+            LOG.error("Error handling worker message type 0x{}", Integer.toHexString(msgType & 0xFF), e);
+        }
+    }
+
+    private void handleWorkerReady(byte[] identity, String identityKey, byte[] payload) throws Exception {
+        WorkerReady ready = WorkerReady.parseFrom(payload);
+        Set<String> exchanges = new HashSet<>(ready.getExchangesList());
+
+        WorkerInfo worker = knownWorkers.computeIfAbsent(identityKey,
+                k -> new WorkerInfo(identity, identityKey, exchanges));
+        worker.exchanges = exchanges; // update in case re-READY with different config
+        worker.identity = identity;
+
+        if (!freeWorkers.contains(worker)) {
+            freeWorkers.addLast(worker);
+        }
+        LOG.info("Ingestor READY: id={}, exchanges={}, freeWorkers={}", identityKey, exchanges, freeWorkers.size());
+
+        dispatchPending();
+    }
+
+    private void handleWorkComplete(String identityKey, byte[] payload) throws Exception {
+        WorkComplete complete = WorkComplete.parseFrom(payload);
+        String jobId = complete.getJobId();
+
+        ActiveJob job = activeJobs.remove(jobId);
+        if (job == null) {
+            LOG.warn("COMPLETE for unknown jobId={}", jobId);
+        } else {
+            LOG.info("Job COMPLETE: jobId={}, ticker={}, success={}", jobId, job.ticker, complete.getSuccess());
+        }
+
+        // Worker is free again
+        WorkerInfo worker = knownWorkers.get(identityKey);
+        if (worker != null) {
+            freeWorkers.addLast(worker);
+            dispatchPending();
+        }
+    }
+
+    private void handleWorkHeartbeat(String identityKey, byte[] payload) throws Exception {
+        WorkHeartbeat hb = WorkHeartbeat.parseFrom(payload);
+        String jobId = hb.getJobId();
+
+        ActiveJob job = activeJobs.get(jobId);
+        if (job != null) {
+            job.lastHeartbeat = System.currentTimeMillis();
+        } else {
+            LOG.warn("HEARTBEAT for unknown jobId={} from worker={}", jobId, identityKey);
+        }
+    }
+
+    private void handleWorkReject(String identityKey, byte[] payload) throws Exception {
+        WorkReject reject = WorkReject.parseFrom(payload);
+        String jobId = reject.getJobId();
+        LOG.warn("Job REJECTED by worker={}: jobId={}, reason={}", identityKey, jobId, reject.getReason());
+
+        ActiveJob job = activeJobs.remove(jobId);
+        if (job != null) {
+            // Re-queue with fresh job_id so a different ingestor may pick it up
+            DataRequest requeued = job.request.toBuilder()
+                    .setJobId(UUID.randomUUID().toString())
+                    .build();
+            pendingJobs.add(requeued);
+        }
+
+        // Worker is still free (it rejected, not crashed)
+        WorkerInfo worker = knownWorkers.get(identityKey);
+        if (worker != null) {
+            freeWorkers.addLast(worker);
+            dispatchPending();
+        }
+    }
+
+    // ── Dispatch ─────────────────────────────────────────────────────────────
+
+    private void enqueueJob(DataRequest request) {
+        // Check if we can immediately dispatch
+        WorkerInfo worker = findFreeWorker(exchangeOf(request.getTicker()));
+        if (worker != null) {
+            dispatch(worker, request);
+        } else {
+            pendingJobs.add(request);
+            LOG.debug("No free worker for {}, queued (pendingJobs={})", request.getTicker(), pendingJobs.size());
+        }
+    }
+
+    private void dispatchPending() {
+        Queue<DataRequest> remaining = new ArrayDeque<>();
+        DataRequest job;
+        while ((job = pendingJobs.poll()) != null) {
+            WorkerInfo worker = findFreeWorker(exchangeOf(job.getTicker()));
+            if (worker != null) {
+                dispatch(worker, job);
+            } else {
+                remaining.add(job);
+            }
+        }
+        pendingJobs.addAll(remaining);
+    }
+
+    private void dispatch(WorkerInfo worker, DataRequest request) {
+        freeWorkers.remove(worker);
+
+        try {
+            byte[] protoBytes = request.toByteArray();
+            boolean sent = zmqManager.sendToWorker(worker.identity, PROTOCOL_VERSION, MSG_TYPE_WORK_ASSIGN, protoBytes);
+            if (!sent) {
+                LOG.error("Failed to dispatch job to worker={}, re-queuing", worker.identityKey);
+                freeWorkers.addLast(worker);
+                pendingJobs.add(request);
+                return;
+            }
+
+            ActiveJob active = new ActiveJob(worker.identity, worker.identityKey,
+                    request, request.getTicker(), request.getType());
+            activeJobs.put(request.getJobId(), active);
+
+            LOG.info("Dispatched job: jobId={}, ticker={}, type={}, worker={}",
+                    request.getJobId(), request.getTicker(), request.getType(), worker.identityKey);
+        } catch (Exception e) {
+            LOG.error("Error dispatching job", e);
+            freeWorkers.addLast(worker);
+        }
+    }
+
+    private void sendStop(byte[] workerIdentity, String jobId) {
+        try {
+            WorkStop stop = WorkStop.newBuilder().setJobId(jobId).build();
+            zmqManager.sendToWorker(workerIdentity, PROTOCOL_VERSION, MSG_TYPE_WORK_STOP, stop.toByteArray());
+        } catch (Exception e) {
+            LOG.error("Error sending STOP for jobId={}", jobId, e);
+        }
+    }
+
+    // ── Timeout checking ─────────────────────────────────────────────────────
+
+    private void checkTimeouts() {
+        long now = System.currentTimeMillis();
+        List<String> timedOut = new ArrayList<>();
+
+        for (Map.Entry<String, ActiveJob> entry : activeJobs.entrySet()) {
+            ActiveJob job = entry.getValue();
+            long timeout = job.type == DataRequest.RequestType.REALTIME_TICKS
+                    ? HEARTBEAT_TIMEOUT_MS : HISTORICAL_TIMEOUT_MS;
+            if (now - job.lastHeartbeat > timeout) {
+                timedOut.add(entry.getKey());
+            }
+        }
+
+        for (String jobId : timedOut) {
+            ActiveJob job = activeJobs.remove(jobId);
+            if (job == null) continue;
+            LOG.warn("Job timed out (no heartbeat/completion): jobId={}, ticker={}, type={}, worker={}",
+                    jobId, job.ticker, job.type, job.workerIdentityKey);
+
+            // Re-queue with a new job_id
+            DataRequest requeued = job.request.toBuilder()
+                    .setJobId(UUID.randomUUID().toString())
+                    .build();
+            pendingJobs.add(requeued);
+            dispatchPending();
+        }
+    }
+
+    // ── Helpers ──────────────────────────────────────────────────────────────
+
+    /** Extract exchange name from ticker, e.g. "BTC/USDT.BINANCE" → "BINANCE" */
+    private static String exchangeOf(String ticker) {
+        int dot = ticker.lastIndexOf('.');
+        return dot >= 0 ? ticker.substring(dot + 1).toUpperCase() : "";
+    }
+
+    /** Find and remove a free worker that supports the given exchange. */
+    private WorkerInfo findFreeWorker(String exchange) {
+        for (WorkerInfo w : freeWorkers) {
+            if (exchange.isEmpty() || w.exchanges.contains(exchange)) {
+                freeWorkers.remove(w);
+                return w;
+            }
+        }
+        return null;
+    }
+
+    private static String bytesToHex(byte[] bytes) {
+        StringBuilder sb = new StringBuilder();
+        for (byte b : bytes) sb.append(String.format("%02x", b));
+        return sb.toString();
+    }
+
+    @Override
+    public void close() {
+        stop();
+    }
+
+    // ── Inner types ──────────────────────────────────────────────────────────
+
+    private static class WorkerInfo {
+        byte[] identity;
+        final String identityKey;
+        Set<String> exchanges;
+
+        WorkerInfo(byte[] identity, String identityKey, Set<String> exchanges) {
+            this.identity = identity;
+            this.identityKey = identityKey;
+            this.exchanges = exchanges;
+        }
+    }
+
+    private static class ActiveJob {
+        final byte[] workerIdentity;
+        final String workerIdentityKey;
+        final DataRequest request;
+        final String ticker;
+        final DataRequest.RequestType type;
+        long lastHeartbeat;
+
+        ActiveJob(byte[] workerIdentity, String workerIdentityKey,
+                  DataRequest request, String ticker, DataRequest.RequestType type) {
+            this.workerIdentity = workerIdentity;
+            this.workerIdentityKey = workerIdentityKey;
+            this.request = request;
+            this.ticker = ticker;
+            this.type = type;
+            this.lastHeartbeat = System.currentTimeMillis();
+        }
+    }
+}

flink/src/main/java/com/dexorder/flink/ingestor/IngestorWorkQueue.java

@@ -119,7 +119,7 @@ public class IngestorWorkQueue {
             String exchangePrefix = extractExchangePrefix(ticker);
 
             boolean sent = zmqManager.sendTopicMessage(
-                ZmqChannelManager.Channel.INGESTOR_WORK_QUEUE,
+                ZmqChannelManager.Channel.INGESTOR_BROKER,
                 exchangePrefix,
                 PROTOCOL_VERSION,
                 MSG_TYPE_DATA_REQUEST,

flink/src/main/java/com/dexorder/flink/ingestor/RealtimeSubscriptionManager.java

@@ -0,0 +1,204 @@
+package com.dexorder.flink.ingestor;
+
+import com.dexorder.flink.zmq.ZmqChannelManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.zeromq.ZMQ;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Monitors XPUB subscription events from the relay and manages realtime ingestor lifecycle.
+ *
+ * This class is the <em>sole owner</em> of the MARKET_DATA_PUB XPUB socket. All outbound
+ * publishes from other threads (e.g., HistoryNotificationForwarder, RealtimeOHLCPublisher)
+ * must go through {@link #enqueuePublish(byte[]...)} so they are sent from the single loop
+ * thread — ZMQ sockets are not thread-safe.
+ *
+ * Topic format: {@code {ticker}|ohlc:{period_seconds}}
+ * Example:      {@code BTC/USDT.BINANCE|ohlc:60}
+ *
+ * Reference counting:
+ *   tickerRefs — across all periods for a ticker; 0→1 triggers ingestor activation
+ *   topicRefs  — per (ticker, period); consulted by RealtimeOHLCPublisher to filter output
+ */
+public class RealtimeSubscriptionManager implements AutoCloseable {
+    private static final Logger LOG = LoggerFactory.getLogger(RealtimeSubscriptionManager.class);
+
+    private static final Pattern TOPIC_PATTERN = Pattern.compile("^(.+)\\|ohlc:(\\d+)$");
+
+    private final ZmqChannelManager zmqManager;
+    private final ZMQ.Socket xpubSocket;
+    private final IngestorBroker broker;
+
+    /** Per-ticker reference count (across all subscribed periods for that ticker) */
+    private final Map<String, Integer> tickerRefs = new HashMap<>();
+
+    /** Per-topic reference count (ticker|ohlc:period → subscriber count) */
+    private final Map<String, Integer> topicRefs = new HashMap<>();
+
+    /**
+     * Thread-safe outbound publish queue.
+     * Each entry is one multi-frame message: {@code byte[][] frames}.
+     */
+    private final ConcurrentLinkedQueue<byte[][]> publishQueue = new ConcurrentLinkedQueue<>();
+
+    private volatile boolean running;
+    private Thread thread;
+
+    public RealtimeSubscriptionManager(ZmqChannelManager zmqManager, IngestorBroker broker) {
+        this.zmqManager = zmqManager;
+        this.xpubSocket = zmqManager.getSocket(ZmqChannelManager.Channel.MARKET_DATA_PUB);
+        this.broker = broker;
+    }
+
+    /**
+     * Queue a multi-frame message for publication on MARKET_DATA_PUB.
+     * Thread-safe — may be called from any thread (HistoryNotificationForwarder,
+     * RealtimeOHLCPublisher, etc.).
+     */
+    public void enqueuePublish(byte[]... frames) {
+        publishQueue.add(frames);
+    }
+
+    /**
+     * Returns the current subscriber count for a topic.
+     * Thread-safe for reads (value is written only from the loop thread but read from others).
+     */
+    public int getTopicRefCount(String topic) {
+        return topicRefs.getOrDefault(topic, 0);
+    }
+
+    public void start() {
+        if (running) {
+            LOG.warn("RealtimeSubscriptionManager already running");
+            return;
+        }
+        running = true;
+        thread = new Thread(this::subscriptionLoop, "RealtimeSubscriptionManager");
+        thread.setDaemon(false);
+        thread.start();
+        LOG.info("RealtimeSubscriptionManager started");
+    }
+
+    public void stop() {
+        running = false;
+        if (thread != null) {
+            thread.interrupt();
+            try {
+                thread.join(5000);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        LOG.info("RealtimeSubscriptionManager stopped");
+    }
+
+    private void subscriptionLoop() {
+        // Build a poller so we can block-wait rather than busy-spin
+        ZMQ.Poller poller = zmqManager.createPoller(1);
+        poller.register(xpubSocket, ZMQ.Poller.POLLIN);
+
+        LOG.info("RealtimeSubscriptionManager loop running");
+
+        while (running) {
+            try {
+                // 1. Flush any queued outbound messages before blocking
+                byte[][] frames;
+                while ((frames = publishQueue.poll()) != null) {
+                    sendFrames(frames);
+                }
+
+                // 2. Wait up to 50ms for a subscription event
+                poller.poll(50);
+
+                // 3. Drain all available subscription events
+                if (poller.pollin(0)) {
+                    byte[] event;
+                    while ((event = xpubSocket.recv(ZMQ.DONTWAIT)) != null) {
+                        if (event.length > 0) {
+                            processSubscriptionEvent(event);
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                if (running) {
+                    LOG.error("Error in subscription loop", e);
+                }
+            }
+        }
+
+        LOG.info("RealtimeSubscriptionManager loop exited");
+    }
+
+    private void sendFrames(byte[][] frames) {
+        for (int i = 0; i < frames.length; i++) {
+            if (i < frames.length - 1) {
+                xpubSocket.sendMore(frames[i]);
+            } else {
+                xpubSocket.send(frames[i], 0);
+            }
+        }
+    }
+
+    private void processSubscriptionEvent(byte[] event) {
+        // XPUB subscription frame: first byte is 0x01 (subscribe) or 0x00 (unsubscribe);
+        // remaining bytes are the raw topic string.
+        boolean isSubscribe = event[0] == 0x01;
+        String topic = new String(event, 1, event.length - 1, ZMQ.CHARSET);
+
+        Matcher m = TOPIC_PATTERN.matcher(topic);
+        if (!m.matches()) {
+            // Not a realtime OHLC topic — e.g. RESPONSE: or HISTORY_READY: prefixes
+            LOG.debug("Ignoring subscription event for non-realtime topic: action={}, topic={}",
+                    isSubscribe ? "subscribe" : "unsubscribe", topic);
+            return;
+        }
+
+        String ticker = m.group(1);
+        LOG.info("Subscription event: action={}, topic={}", isSubscribe ? "subscribe" : "unsubscribe", topic);
+
+        if (isSubscribe) {
+            handleSubscribe(ticker, topic);
+        } else {
+            handleUnsubscribe(ticker, topic);
+        }
+    }
+
+    private void handleSubscribe(String ticker, String topic) {
+        int newTopicRef = topicRefs.merge(topic, 1, Integer::sum);
+        LOG.debug("topicRefs[{}]={}", topic, newTopicRef);
+
+        int newTickerRef = tickerRefs.merge(ticker, 1, Integer::sum);
+        if (newTickerRef == 1) {
+            LOG.info("First subscriber for ticker={} — submitting realtime request", ticker);
+            broker.submitRealtimeRequest(ticker);
+        }
+        LOG.debug("tickerRefs[{}]={}", ticker, newTickerRef);
+    }
+
+    private void handleUnsubscribe(String ticker, String topic) {
+        int newTopicRef = topicRefs.merge(topic, -1, Integer::sum);
+        if (newTopicRef <= 0) {
+            topicRefs.remove(topic);
+        }
+        LOG.debug("topicRefs[{}]={}", topic, newTopicRef);
+
+        int newTickerRef = tickerRefs.merge(ticker, -1, Integer::sum);
+        if (newTickerRef <= 0) {
+            tickerRefs.remove(ticker);
+            LOG.info("Last subscriber for ticker={} left — stopping realtime jobs", ticker);
+            broker.stopRealtimeJobsForTicker(ticker);
+        }
+        LOG.debug("tickerRefs[{}]={}", ticker, newTickerRef);
+    }
+
+    @Override
+    public void close() {
+        stop();
+    }
+}

flink/src/main/java/com/dexorder/flink/publisher/HistoryNotificationForwarder.java

@@ -6,14 +6,24 @@ import org.zeromq.SocketType;
 import org.zeromq.ZContext;
 import org.zeromq.ZMQ;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+
 /**
  * Runs in the job manager. Pulls notifications from task managers (via PUSH/PULL)
- * and republishes them on the MARKET_DATA_PUB socket that the relay subscribes to.
+ * and enqueues them for publication on MARKET_DATA_PUB via the provided publish callback.
+ *
+ * The publish callback must be thread-safe (e.g., RealtimeSubscriptionManager.enqueuePublish).
+ * Direct socket access is avoided here because the MARKET_DATA_PUB XPUB socket is owned
+ * exclusively by RealtimeSubscriptionManager to satisfy ZMQ's single-thread-per-socket rule.
  *
  * Flow:
  *   Task manager HistoryNotificationPublisher  →  PUSH
  *                                               ↓
- *   Job manager HistoryNotificationForwarder   PULL  →  MARKET_DATA_PUB
+ *   Job manager HistoryNotificationForwarder   PULL  →  publishCallback (queue)
+ *                                                        ↓ (RealtimeSubscriptionManager loop)
+ *                                                        MARKET_DATA_PUB
  *                                                        ↓
  *   Relay (XSUB) → Relay (XPUB) → Clients
  */
@@ -21,17 +31,17 @@ public class HistoryNotificationForwarder implements AutoCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(HistoryNotificationForwarder.class);
 
     private final ZMQ.Socket pullSocket;
-    private final ZMQ.Socket pubSocket;
+    private final Consumer<byte[][]> publishCallback;
     private final ZContext context;
     private volatile boolean running = true;
     private Thread thread;
 
     /**
-     * @param pullPort   Port to bind PULL socket on (task managers connect PUSH here)
-     * @param pubSocket  Existing MARKET_DATA_PUB socket from ZmqChannelManager
+     * @param pullPort        Port to bind PULL socket on (task managers connect PUSH here)
+     * @param publishCallback Thread-safe callback to enqueue outbound multi-frame messages
      */
-    public HistoryNotificationForwarder(int pullPort, ZMQ.Socket pubSocket) {
-        this.pubSocket = pubSocket;
+    public HistoryNotificationForwarder(int pullPort, Consumer<byte[][]> publishCallback) {
+        this.publishCallback = publishCallback;
         this.context = new ZContext();
         this.pullSocket = context.createSocket(SocketType.PULL);
         this.pullSocket.setRcvHWM(10000);
@@ -53,32 +63,24 @@ public class HistoryNotificationForwarder implements AutoCloseable {
         pullSocket.setReceiveTimeOut(200); // ms, so we can check running flag
 
         while (running) {
-            // Receive all frames of a multi-part message and forward to PUB
             byte[] frame = pullSocket.recv(0);
             if (frame == null) {
-                continue; // timeout, check running flag
+                continue; // timeout — check running flag
             }
 
-            boolean more = pullSocket.hasReceiveMore();
-            if (more) {
-                pubSocket.sendMore(frame);
-            } else {
-                pubSocket.send(frame, 0);
-                continue;
-            }
+            // Collect all frames of the multi-part message, then enqueue atomically
+            List<byte[]> frames = new ArrayList<>();
+            frames.add(frame);
 
-            // Receive remaining frames
-            while (more) {
-                frame = pullSocket.recv(0);
-                more = pullSocket.hasReceiveMore();
-                if (more) {
-                    pubSocket.sendMore(frame);
-                } else {
-                    pubSocket.send(frame, 0);
+            while (pullSocket.hasReceiveMore()) {
+                byte[] next = pullSocket.recv(0);
+                if (next != null) {
+                    frames.add(next);
                 }
             }
 
-            LOG.debug("Forwarded notification to MARKET_DATA_PUB");
+            publishCallback.accept(frames.toArray(new byte[0][]));
+            LOG.debug("Enqueued notification ({} frames) for MARKET_DATA_PUB", frames.size());
         }
 
         LOG.info("Notification forwarder loop stopped");

flink/src/main/java/com/dexorder/flink/publisher/HistoryNotificationFunction.java

@@ -64,8 +64,13 @@ public class HistoryNotificationFunction extends ProcessFunction<OHLCBatchWrappe
         String status = batch.getStatus();
         int rowCount = batch.getRowCount();
 
-        LOG.info("Processing OHLCBatch: request_id={}, status={}, rows={}",
-                requestId, status, rowCount);
+        LOG.info("Processing OHLCBatch: request_id={}, status={}, rows={}, isLastPage={}",
+                requestId, status, rowCount, batch.isLastPage());
+
+        // Intermediate pages: data is written to Iceberg but no notification yet
+        if (!batch.isLastPage()) {
+            return;
+        }
 
         // Determine Iceberg table name based on period
         String tableName = getIcebergTableName(ticker, periodSeconds);

flink/src/main/java/com/dexorder/flink/publisher/OHLCBatchDeserializer.java

@@ -87,7 +87,8 @@ public class OHLCBatchDeserializer implements DeserializationSchema<OHLCBatchWra
                 meta.getEndTime(),
                 status,
                 meta.hasErrorMessage() ? meta.getErrorMessage() : null,
-                rows
+                rows,
+                meta.getIsLastPage()
         );
     }
 

flink/src/main/java/com/dexorder/flink/publisher/OHLCBatchWrapper.java

@@ -19,6 +19,7 @@ public class OHLCBatchWrapper implements Serializable {
     private final String status; // OK, NOT_FOUND, ERROR
     private final String errorMessage;
     private final List<OHLCRow> rows;
+    private final boolean isLastPage;
 
     public OHLCBatchWrapper(
             String requestId,
@@ -29,7 +30,8 @@ public class OHLCBatchWrapper implements Serializable {
             long endTime,
             String status,
             String errorMessage,
-            List<OHLCRow> rows
+            List<OHLCRow> rows,
+            boolean isLastPage
     ) {
         this.requestId = requestId;
         this.clientId = clientId;
@@ -40,6 +42,7 @@ public class OHLCBatchWrapper implements Serializable {
         this.status = status;
         this.errorMessage = errorMessage;
         this.rows = rows;
+        this.isLastPage = isLastPage;
     }
 
     public String getRequestId() {
@@ -94,6 +97,10 @@ public class OHLCBatchWrapper implements Serializable {
         return "OK".equals(status);
     }
 
+    public boolean isLastPage() {
+        return isLastPage;
+    }
+
     @Override
     public String toString() {
         return "OHLCBatchWrapper{" +
@@ -103,6 +110,7 @@ public class OHLCBatchWrapper implements Serializable {
                 ", periodSeconds=" + periodSeconds +
                 ", status='" + status + '\'' +
                 ", rowCount=" + getRowCount() +
+                ", isLastPage=" + isLastPage +
                 '}';
     }
 

flink/src/main/java/com/dexorder/flink/publisher/RealtimeBar.java

@@ -0,0 +1,73 @@
+package com.dexorder.flink.publisher;
+
+import java.io.Serializable;
+
+/**
+ * A single completed OHLC bar for a given ticker and period.
+ * Output type of RealtimeBarFunction, input type of RealtimeBarPublisher.
+ */
+public class RealtimeBar implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private String ticker;
+    /** Period in seconds (e.g., 60, 300, 3600) */
+    private int periodSeconds;
+    /** Window start timestamp in milliseconds since epoch */
+    private long windowStartMs;
+    /** Scaled integer price values (same precision as source Tick) */
+    private long open;
+    private long high;
+    private long low;
+    private long close;
+    /** Summed base amount across ticks in this window */
+    private long volume;
+    /** Number of ticks in this window */
+    private int tickCount;
+
+    public RealtimeBar() {}
+
+    public RealtimeBar(String ticker, int periodSeconds, long windowStartMs,
+                       long open, long high, long low, long close, long volume, int tickCount) {
+        this.ticker = ticker;
+        this.periodSeconds = periodSeconds;
+        this.windowStartMs = windowStartMs;
+        this.open = open;
+        this.high = high;
+        this.low = low;
+        this.close = close;
+        this.volume = volume;
+        this.tickCount = tickCount;
+    }
+
+    public String getTicker() { return ticker; }
+    public int getPeriodSeconds() { return periodSeconds; }
+    public long getWindowStartMs() { return windowStartMs; }
+    public long getOpen() { return open; }
+    public long getHigh() { return high; }
+    public long getLow() { return low; }
+    public long getClose() { return close; }
+    public long getVolume() { return volume; }
+    public int getTickCount() { return tickCount; }
+
+    public void setTicker(String ticker) { this.ticker = ticker; }
+    public void setPeriodSeconds(int periodSeconds) { this.periodSeconds = periodSeconds; }
+    public void setWindowStartMs(long windowStartMs) { this.windowStartMs = windowStartMs; }
+    public void setOpen(long open) { this.open = open; }
+    public void setHigh(long high) { this.high = high; }
+    public void setLow(long low) { this.low = low; }
+    public void setClose(long close) { this.close = close; }
+    public void setVolume(long volume) { this.volume = volume; }
+    public void setTickCount(int tickCount) { this.tickCount = tickCount; }
+
+    /** ZMQ topic for this bar: e.g., "BTC/USDT.BINANCE|ohlc:60" */
+    public String topic() {
+        return ticker + "|ohlc:" + periodSeconds;
+    }
+
+    @Override
+    public String toString() {
+        return "RealtimeBar{ticker='" + ticker + "', period=" + periodSeconds +
+                "s, windowStart=" + windowStartMs + ", O=" + open + " H=" + high +
+                " L=" + low + " C=" + close + ", ticks=" + tickCount + '}';
+    }
+}

flink/src/main/java/com/dexorder/flink/publisher/RealtimeBarFunction.java

@@ -0,0 +1,116 @@
+package com.dexorder.flink.publisher;
+
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Accumulates ticks into OHLC bars for each configured period.
+ *
+ * Keyed by ticker. Maintains per-period accumulators in MapState.
+ * Uses a "lazy boundary" approach: a new window is detected when a tick arrives after
+ * the previous window's end time (based on processing clock). The completed bar is
+ * emitted immediately when the boundary is crossed, so bars are delayed by at most
+ * one tick interval (~10s for realtime polling).
+ *
+ * Periods are configurable at construction time. All configured periods are computed
+ * for every ticker receiving ticks; the ZMQ publisher filters to active subscriptions.
+ *
+ * Accumulator layout (long[7]):
+ *   [0] open
+ *   [1] high
+ *   [2] low
+ *   [3] close
+ *   [4] volume (sum of base amount)
+ *   [5] windowStartMs (epoch ms)
+ *   [6] tickCount
+ */
+public class RealtimeBarFunction extends RichFlatMapFunction<TickWrapper, RealtimeBar> {
+    private static final Logger LOG = LoggerFactory.getLogger(RealtimeBarFunction.class);
+    private static final long serialVersionUID = 1L;
+
+    private final int[] periods;
+    private transient MapState<Integer, long[]> accumState;
+
+    /**
+     * @param periods  Period lengths in seconds (e.g., 60, 300, 900, 3600)
+     */
+    public RealtimeBarFunction(int[] periods) {
+        this.periods = periods;
+    }
+
+    @Override
+    public void open(Configuration parameters) {
+        MapStateDescriptor<Integer, long[]> desc = new MapStateDescriptor<>(
+                "ohlcAccum",
+                BasicTypeInfo.INT_TYPE_INFO,
+                PrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO
+        );
+        accumState = getRuntimeContext().getMapState(desc);
+    }
+
+    @Override
+    public void flatMap(TickWrapper tick, Collector<RealtimeBar> out) throws Exception {
+        if (tick == null) return;
+
+        long nowMs = System.currentTimeMillis();
+
+        for (int period : periods) {
+            long periodMs = period * 1000L;
+            long windowStart = (nowMs / periodMs) * periodMs;
+
+            long[] accum = accumState.get(period);
+
+            if (accum == null) {
+                // First tick for this period
+                accumState.put(period, openWindow(tick, windowStart));
+
+            } else if (accum[5] != windowStart) {
+                // Window boundary crossed — emit completed bar then start fresh
+                if (accum[6] > 0) {
+                    out.collect(toBar(tick.getTicker(), period, accum));
+                    LOG.debug("Emitted bar: ticker={}, period={}s, windowStart={}, ticks={}",
+                            tick.getTicker(), period, accum[5], accum[6]);
+                }
+                accumState.put(period, openWindow(tick, windowStart));
+
+            } else {
+                // Same window — update
+                accum[1] = Math.max(accum[1], tick.getPrice()); // high
+                accum[2] = Math.min(accum[2], tick.getPrice()); // low
+                accum[3] = tick.getPrice();                      // close
+                accum[4] += tick.getAmount();                    // volume
+                accum[6]++;                                       // tick count
+                accumState.put(period, accum);
+            }
+        }
+    }
+
+    private static long[] openWindow(TickWrapper tick, long windowStart) {
+        return new long[]{
+            tick.getPrice(), // open
+            tick.getPrice(), // high
+            tick.getPrice(), // low
+            tick.getPrice(), // close
+            tick.getAmount(), // volume
+            windowStart,
+            1L               // tickCount
+        };
+    }
+
+    private static RealtimeBar toBar(String ticker, int period, long[] accum) {
+        return new RealtimeBar(
+                ticker, period,
+                accum[5],           // windowStartMs
+                accum[0], accum[1], accum[2], accum[3], // O H L C
+                accum[4],           // volume
+                (int) accum[6]      // tickCount
+        );
+    }
+}

flink/src/main/java/com/dexorder/flink/publisher/RealtimeBarPublisher.java

@@ -0,0 +1,91 @@
+package com.dexorder.flink.publisher;
+
+import com.dexorder.proto.OHLC;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.zeromq.SocketType;
+import org.zeromq.ZContext;
+import org.zeromq.ZMQ;
+
+/**
+ * Flink sink that publishes completed realtime OHLC bars to clients.
+ *
+ * Connects a ZMQ PUSH socket to the job manager's notification PULL endpoint.
+ * The HistoryNotificationForwarder (already running on the job manager) receives these
+ * frames and enqueues them to RealtimeSubscriptionManager, which publishes them on
+ * the MARKET_DATA_PUB XPUB socket. Clients subscribed to the matching topic receive the bar.
+ *
+ * Wire format (matches HistoryNotificationPublisher):
+ *   Frame 1: topic bytes  (e.g., "BTC/USDT.BINANCE|ohlc:60")
+ *   Frame 2: [0x01]       (protocol version)
+ *   Frame 3: [0x04][OHLC protobuf bytes]  (type 0x04 = OHLC single bar)
+ *
+ * Parallelism MUST be 1 (same as the rest of the notification pipeline).
+ */
+public class RealtimeBarPublisher extends RichSinkFunction<RealtimeBar> {
+    private static final Logger LOG = LoggerFactory.getLogger(RealtimeBarPublisher.class);
+    private static final long serialVersionUID = 1L;
+
+    private static final byte PROTOCOL_VERSION = 0x01;
+    private static final byte MSG_TYPE_OHLC = 0x04;
+
+    private final String jobManagerPullEndpoint;
+
+    private transient ZContext context;
+    private transient ZMQ.Socket pushSocket;
+
+    public RealtimeBarPublisher(String jobManagerPullEndpoint) {
+        this.jobManagerPullEndpoint = jobManagerPullEndpoint;
+    }
+
+    @Override
+    public void open(Configuration parameters) {
+        context = new ZContext();
+        pushSocket = context.createSocket(SocketType.PUSH);
+        pushSocket.setLinger(1000);
+        pushSocket.setSndHWM(10000);
+        pushSocket.connect(jobManagerPullEndpoint);
+        LOG.info("RealtimeBarPublisher PUSH connected to {}", jobManagerPullEndpoint);
+    }
+
+    @Override
+    public void invoke(RealtimeBar bar, Context context) {
+        try {
+            // Build OHLC proto — timestamp in nanoseconds (bar uses ms, convert)
+            OHLC ohlc = OHLC.newBuilder()
+                    .setTimestamp(bar.getWindowStartMs() * 1_000_000L) // ms → ns
+                    .setTicker(bar.getTicker())
+                    .setOpen(bar.getOpen())
+                    .setHigh(bar.getHigh())
+                    .setLow(bar.getLow())
+                    .setClose(bar.getClose())
+                    .setVolume(bar.getVolume())
+                    .build();
+
+            byte[] protoBytes = ohlc.toByteArray();
+            byte[] messageFrame = new byte[1 + protoBytes.length];
+            messageFrame[0] = MSG_TYPE_OHLC;
+            System.arraycopy(protoBytes, 0, messageFrame, 1, protoBytes.length);
+
+            String topic = bar.topic();
+            pushSocket.sendMore(topic.getBytes(ZMQ.CHARSET));
+            pushSocket.sendMore(new byte[]{PROTOCOL_VERSION});
+            pushSocket.send(messageFrame, 0);
+
+            LOG.debug("Published realtime bar: topic={}, ticks={}", topic, bar.getTickCount());
+
+        } catch (Exception e) {
+            LOG.error("Failed to publish realtime bar: ticker={}, period={}",
+                    bar.getTicker(), bar.getPeriodSeconds(), e);
+        }
+    }
+
+    @Override
+    public void close() {
+        if (pushSocket != null) pushSocket.close();
+        if (context != null) context.close();
+        LOG.info("RealtimeBarPublisher closed");
+    }
+}

flink/src/main/java/com/dexorder/flink/publisher/TickDeserializer.java

@@ -0,0 +1,69 @@
+package com.dexorder.flink.publisher;
+
+import com.dexorder.proto.Tick;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Kafka deserializer for Tick protobuf messages from the market-tick topic.
+ *
+ * Wire format: [0x01 version][0x03 TICK type][Tick protobuf bytes]
+ */
+public class TickDeserializer implements DeserializationSchema<TickWrapper> {
+    private static final Logger LOG = LoggerFactory.getLogger(TickDeserializer.class);
+    private static final long serialVersionUID = 1L;
+
+    private static final byte PROTOCOL_VERSION = 0x01;
+    private static final byte MSG_TYPE_TICK = 0x03;
+
+    @Override
+    public TickWrapper deserialize(byte[] message) throws IOException {
+        try {
+            if (message.length < 2) {
+                throw new IOException("Message too short: " + message.length + " bytes");
+            }
+
+            if (message[0] != PROTOCOL_VERSION) {
+                throw new IOException("Unsupported protocol version: 0x" + Integer.toHexString(message[0] & 0xFF));
+            }
+
+            if (message[1] != MSG_TYPE_TICK) {
+                throw new IOException("Unexpected message type: 0x" + Integer.toHexString(message[1] & 0xFF));
+            }
+
+            byte[] payload = new byte[message.length - 2];
+            System.arraycopy(message, 2, payload, 0, payload.length);
+
+            Tick tick = Tick.parseFrom(payload);
+
+            return new TickWrapper(
+                    tick.getTicker(),
+                    tick.getTradeId(),
+                    tick.getTimestamp(),
+                    tick.getPrice(),
+                    tick.getAmount(),
+                    tick.getQuoteAmount(),
+                    tick.getTakerBuy()
+            );
+
+        } catch (Exception e) {
+            LOG.warn("Failed to deserialize Tick, skipping: {}", e.getMessage());
+            // Return null; Flink's KafkaSource skips nulls via filter
+            return null;
+        }
+    }
+
+    @Override
+    public boolean isEndOfStream(TickWrapper nextElement) {
+        return false;
+    }
+
+    @Override
+    public TypeInformation<TickWrapper> getProducedType() {
+        return TypeInformation.of(TickWrapper.class);
+    }
+}

flink/src/main/java/com/dexorder/flink/publisher/TickWrapper.java

@@ -0,0 +1,58 @@
+package com.dexorder.flink.publisher;
+
+import java.io.Serializable;
+
+/**
+ * Flink-serializable wrapper for a single Tick.
+ * Fields mirror the Tick protobuf, using primitives to avoid proto-class serialization issues.
+ */
+public class TickWrapper implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private String ticker;
+    private String tradeId;
+    /** Timestamp in nanoseconds since epoch */
+    private long timestamp;
+    /** Price as scaled integer */
+    private long price;
+    /** Base amount as scaled integer */
+    private long amount;
+    /** Quote amount as scaled integer */
+    private long quoteAmount;
+    private boolean takerBuy;
+
+    public TickWrapper() {}
+
+    public TickWrapper(String ticker, String tradeId, long timestamp,
+                       long price, long amount, long quoteAmount, boolean takerBuy) {
+        this.ticker = ticker;
+        this.tradeId = tradeId;
+        this.timestamp = timestamp;
+        this.price = price;
+        this.amount = amount;
+        this.quoteAmount = quoteAmount;
+        this.takerBuy = takerBuy;
+    }
+
+    public String getTicker() { return ticker; }
+    public String getTradeId() { return tradeId; }
+    public long getTimestamp() { return timestamp; }
+    public long getPrice() { return price; }
+    public long getAmount() { return amount; }
+    public long getQuoteAmount() { return quoteAmount; }
+    public boolean isTakerBuy() { return takerBuy; }
+
+    public void setTicker(String ticker) { this.ticker = ticker; }
+    public void setTradeId(String tradeId) { this.tradeId = tradeId; }
+    public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
+    public void setPrice(long price) { this.price = price; }
+    public void setAmount(long amount) { this.amount = amount; }
+    public void setQuoteAmount(long quoteAmount) { this.quoteAmount = quoteAmount; }
+    public void setTakerBuy(boolean takerBuy) { this.takerBuy = takerBuy; }
+
+    @Override
+    public String toString() {
+        return "TickWrapper{ticker='" + ticker + "', tradeId='" + tradeId +
+                "', timestamp=" + timestamp + ", price=" + price + '}';
+    }
+}

flink/src/main/java/com/dexorder/flink/zmq/ZmqChannelManager.java

@@ -13,7 +13,14 @@ import java.util.Map;
 
 /**
  * Manages all ZeroMQ channels for the Flink application.
- * Each channel is bound to a specific port and socket type.
+ *
+ * Port layout:
+ *   5558  XPUB  MARKET_DATA_PUB   — market data + notifications to clients (via relay XSUB)
+ *                                   XPUB exposes subscription frames so Flink can detect
+ *                                   which realtime topics clients are interested in.
+ *   5561  PULL  (internal)         — task manager → job manager notifications (unchanged)
+ *   5566  PULL  CLIENT_REQUEST     — receives forwarded SubmitHistoricalRequest from relay PUSH
+ *   5567  ROUTER INGESTOR_BROKER   — exclusive work queue; ingestors connect with DEALER
  */
 public class ZmqChannelManager implements Closeable {
     private static final Logger LOG = LoggerFactory.getLogger(ZmqChannelManager.class);
@@ -23,8 +30,9 @@ public class ZmqChannelManager implements Closeable {
     private final AppConfig config;
 
     public enum Channel {
-        INGESTOR_WORK_QUEUE,
         MARKET_DATA_PUB,
+        CLIENT_REQUEST,
+        INGESTOR_BROKER,
     }
 
     public ZmqChannelManager(AppConfig config) {
@@ -41,20 +49,33 @@ public class ZmqChannelManager implements Closeable {
 
         LOG.info("Initializing ZeroMQ channels on {}", bindAddress);
 
-        // 1. Ingestor Work Queue - PUB socket for topic-based work distribution (exchange prefix filtering)
+        // 1. Market Data Publication — XPUB so subscription events are visible to Flink
+        //    Relay's XSUB connects here to proxy data to clients.
+        //    Subscription frames from relay (forwarded from clients) arrive as readable messages.
+        ZMQ.Socket marketDataSocket = context.createSocket(SocketType.XPUB);
+        marketDataSocket.setXpubVerbose(true);   // emit every sub/unsub, not just first/last
+        marketDataSocket.setLinger(1000);
+        marketDataSocket.setSndHWM(10000);
+        marketDataSocket.setRcvHWM(10000);
+        String marketDataEndpoint = bindAddress + ":" + config.getMarketDataPubPort();
+        marketDataSocket.bind(marketDataEndpoint);
+        sockets.put(Channel.MARKET_DATA_PUB.name(), marketDataSocket);
+        LOG.info("Bound Market Data Publication (XPUB) to {}", marketDataEndpoint);
+
+        // 2. Client Request Pull — receives SubmitHistoricalRequest forwarded by relay PUSH
         createAndBind(
-            Channel.INGESTOR_WORK_QUEUE,
-            SocketType.PUB,
-            bindAddress + ":" + config.getIngestorWorkQueuePort(),
-            "Ingestor Work Queue (PUB)"
+            Channel.CLIENT_REQUEST,
+            SocketType.PULL,
+            bindAddress + ":" + config.getFlinkRequestPullPort(),
+            "Client Request (PULL)"
         );
 
-        // 2. Market Data Publication - PUB socket for market data streaming and HistoryReadyNotification
+        // 3. Ingestor Broker — ROUTER for exclusive work dispatch to ingestor DEALER workers
         createAndBind(
-            Channel.MARKET_DATA_PUB,
-            SocketType.PUB,
-            bindAddress + ":" + config.getMarketDataPubPort(),
-            "Market Data Publication (PUB)"
+            Channel.INGESTOR_BROKER,
+            SocketType.ROUTER,
+            bindAddress + ":" + config.getIngestorBrokerPort(),
+            "Ingestor Broker (ROUTER)"
         );
 
         LOG.info("All ZeroMQ channels initialized successfully");
@@ -63,15 +84,10 @@ public class ZmqChannelManager implements Closeable {
     private void createAndBind(Channel channel, SocketType socketType, String endpoint, String description) {
         try {
             ZMQ.Socket socket = context.createSocket(socketType);
-
-            // Set socket options
-            socket.setLinger(1000); // 1 second linger on close
-            socket.setSndHWM(10000); // High water mark for outbound messages
-            socket.setRcvHWM(10000); // High water mark for inbound messages
-
-            // Bind the socket
+            socket.setLinger(1000);
+            socket.setSndHWM(10000);
+            socket.setRcvHWM(10000);
             socket.bind(endpoint);
-
             sockets.put(channel.name(), socket);
             LOG.info("Bound {} to {}", description, endpoint);
         } catch (Exception e) {
@@ -80,6 +96,13 @@ public class ZmqChannelManager implements Closeable {
         }
     }
 
+    /**
+     * Create a ZMQ Poller backed by this manager's context.
+     */
+    public ZMQ.Poller createPoller(int size) {
+        return context.getContext().poller(size);
+    }
+
     /**
      * Get a socket by channel type.
      */
@@ -92,18 +115,11 @@ public class ZmqChannelManager implements Closeable {
     }
 
     /**
-     * Send a message on the specified channel.
-     *
-     * @param channel The channel to send on
-     * @param versionByte Protocol version byte
-     * @param messageTypeByte Message type ID byte
-     * @param protobufData Serialized protobuf message
-     * @return true if sent successfully
+     * Send a message on a channel (no topic prefix — for PULL/PUSH or direct sends).
      */
     public boolean sendMessage(Channel channel, byte versionByte, byte messageTypeByte, byte[] protobufData) {
         ZMQ.Socket socket = getSocket(channel);
 
-        // Send as two frames: [version byte] [type byte + protobuf data]
         boolean sentFrame1 = socket.send(new byte[]{versionByte}, ZMQ.SNDMORE);
         if (!sentFrame1) {
             LOG.error("Failed to send version frame on channel {}", channel);
@@ -124,27 +140,18 @@ public class ZmqChannelManager implements Closeable {
     }
 
     /**
-     * Send a message with a topic prefix (for PUB sockets).
-     *
-     * @param channel The channel to send on
-     * @param topic Topic string for subscription filtering
-     * @param versionByte Protocol version byte
-     * @param messageTypeByte Message type ID byte
-     * @param protobufData Serialized protobuf message
-     * @return true if sent successfully
+     * Send a topic-prefixed message (for XPUB market data publishing).
+     * Frame layout: [topic][version][type+payload]
      */
     public boolean sendTopicMessage(Channel channel, String topic, byte versionByte, byte messageTypeByte, byte[] protobufData) {
         ZMQ.Socket socket = getSocket(channel);
 
-        // Send as three frames: [topic] [version byte] [type byte + protobuf data]
-        boolean sentTopic = socket.send(topic.getBytes(ZMQ.CHARSET), ZMQ.SNDMORE);
-        if (!sentTopic) {
+        if (!socket.send(topic.getBytes(ZMQ.CHARSET), ZMQ.SNDMORE)) {
             LOG.error("Failed to send topic frame on channel {}", channel);
             return false;
         }
 
-        boolean sentFrame1 = socket.send(new byte[]{versionByte}, ZMQ.SNDMORE);
-        if (!sentFrame1) {
+        if (!socket.send(new byte[]{versionByte}, ZMQ.SNDMORE)) {
             LOG.error("Failed to send version frame on channel {}", channel);
             return false;
         }
@@ -153,8 +160,7 @@ public class ZmqChannelManager implements Closeable {
         frame2[0] = messageTypeByte;
         System.arraycopy(protobufData, 0, frame2, 1, protobufData.length);
 
-        boolean sentFrame2 = socket.send(frame2, 0);
-        if (!sentFrame2) {
+        if (!socket.send(frame2, 0)) {
             LOG.error("Failed to send message frame on channel {}", channel);
             return false;
         }
@@ -162,6 +168,24 @@ public class ZmqChannelManager implements Closeable {
         return true;
     }
 
+    /**
+     * Send a targeted message to a specific DEALER worker via ROUTER.
+     * Frame layout: [identity][empty][version][type+payload]
+     */
+    public boolean sendToWorker(byte[] identity, byte versionByte, byte messageTypeByte, byte[] protobufData) {
+        ZMQ.Socket socket = getSocket(Channel.INGESTOR_BROKER);
+
+        if (!socket.send(identity, ZMQ.SNDMORE)) return false;
+        if (!socket.send(new byte[0], ZMQ.SNDMORE)) return false;
+        if (!socket.send(new byte[]{versionByte}, ZMQ.SNDMORE)) return false;
+
+        byte[] frame = new byte[1 + protobufData.length];
+        frame[0] = messageTypeByte;
+        System.arraycopy(protobufData, 0, frame, 1, protobufData.length);
+
+        return socket.send(frame, 0);
+    }
+
     @Override
     public void close() {
         LOG.info("Closing ZeroMQ channels");