Skip to content

gRPC Push Repository Connector

Jump to bottom
Piergiorgio Lucidi edited this page Oct 19, 2026 · 5 revisions

gRPC Push Repository Connector

gRPC Protobuf Java Spring Boot OIS Compliance Claim Check OpenTelemetry

Module: oc-grpc-push-connector
Connector Class: org.opencrawling.push.grpc.GrpcPushRepositoryConnector
gRPC Service Implementation: org.opencrawling.push.grpc.OisPushServiceImpl
Server Lifecycle Manager: org.opencrawling.push.grpc.GrpcPushServerManager
Document Translator & Claim-Check Offloader: org.opencrawling.push.grpc.OisDocumentTranslator
Runtime Dispatcher: org.opencrawling.runtime.transport.RuntimePushDocumentDispatcher
Default Port: 9096

📌 Overview

The gRPC Push Repository Connector (oc-grpc-push-connector) provides OpenCrawling with a high-throughput, sub-millisecond, binary streaming ingestion gateway built on HTTP/2 and Protocol Buffers v3.

While traditional repository connectors operate on a pull/poll paradigm—periodically scanning file systems, ECM repositories, or relational databases—modern event-driven architectures often require push-based ingestion. Change Data Capture (CDC) pipelines (Debezium, Kafka Connect), enterprise webhook listeners, custom microservices, and distributed web crawler bolts need to push documents and real-time lifecycle events (UPSERT and DELETE) directly into OpenCrawling as they happen.

By combining gRPC HTTP/2 multiplexing, Java 25 Virtual Threads, automatic Claim-Check Pattern offloading (Apache Ozone, AWS S3, or local storage), Zero-Trust Security & ACL mapping, W3C Trace Context propagation, and TLS / mTLS mutual authentication, the gRPC Push Repository Connector turns OpenCrawling into a universal, language-agnostic streaming ingestion endpoint for the Open Ingestion Standard (OIS).

flowchart TD
    subgraph Producers["External Event Producers"]
        CDC["Debezium / CDC Pipelines"]
        Webhook["Enterprise Webhook Listeners"]
        Micro["Polyglot Microservices: Go / Rust / Python / Java"]
        Crawler["Distributed Crawler Workers"]
        S3Direct["Direct S3 / Ozone Pre-Uploader"]
    end

    subgraph GrpcPushConnector["oc-grpc-push-connector (Port 9096 - Java 25 Virtual Threads)"]
        Netty["Shaded Netty HTTP/2 Server - TLS / mTLS"]
        SecInt["GrpcSecurityInterceptor - Bearer Token Auth"]
        OTelInt["OpenTelemetryServerInterceptor - W3C traceparent"]
        Service["OisPushServiceImpl: Unary / Client Stream / BiDi Stream / Ping"]
        Translator["OisDocumentTranslator: ACLs & Claim-Check Threshold Check"]

        CDC -->|gRPC StreamDocuments| Netty
        Webhook -->|gRPC PushDocument| Netty
        Micro -->|gRPC BiDirectionalStream| Netty
        Crawler -->|gRPC PushDocument| Netty
        
        Netty --> SecInt
        SecInt --> OTelInt
        OTelInt --> Service
        Service --> Translator
    end

    subgraph ClaimCheckTier["Claim-Check Storage Tier"]
        ClaimStore[("ClaimCheckStore: Local / Apache Ozone / AWS S3")]
        S3Direct -.->|1. Pre-Upload Multi-GB Binary| ClaimStore
        S3Direct -->|2. PushDocument with claim_check_uri| Netty
        Translator -.->|Offload if payload > inlineBinaryThresholdKb| ClaimStore
    end

    subgraph Runtime["OpenCrawling Ingestion Runtime (oc-runtime)"]
        Dispatcher["RuntimePushDocumentDispatcher"]
        Orchestrator["JobOrchestrator"]
        
        Translator -->|RepositoryDocument| Dispatcher
        Dispatcher -->|Validate Job & PipelineMode| Orchestrator
        
        Orchestrator -->|Action == DELETE: Tombstone Short-Circuit| WriterTopic[("Kafka: opencrawling-embedded")]
        Orchestrator -->|Action == UPSERT: IngestionMessage| IngestTopic[("Kafka: opencrawling-ingestion")]
        
        IngestTopic --> IngestionConsumer["IngestionConsumer: Apache Tika 4 + TokenTextSplitter"]
        IngestionConsumer -.->|Read Offloaded Binary| ClaimStore
        IngestionConsumer --> Embed["Embedding Service: Ollama / OpenAI"]
        Embed --> WriterTopic
        
        WriterTopic --> Sinks[("Vector Stores: pgvector / Milvus / Qdrant / Vespa / OpenSearch / Solr")]
        Sinks --> MCP["Secure MCP Server - Zero-Trust ACL Enforcement"]
    end
Loading

🚀 Key Features

  • Four Native gRPC Interaction Models:
    • Unary (PushDocument): Request-response ingestion for single documents, transactional webhooks, or real-time updates with immediate ingestion status feedback.
    • Client Streaming (StreamDocuments): High-throughput bulk ingestion stream where the client pushes a continuous sequence of documents and receives an aggregated StreamSummaryResponse (accepted_count, rejected_count, and per-document PushFailure diagnostics).
    • Bi-Directional Streaming (BiDirectionalStream): Full-duplex asynchronous streaming where the client streams PushDocumentRequest messages and receives real-time per-document IngestionAck frames (correlation_id, document_id, status, server_timestamp) with reactive backpressure.
    • Diagnostic Health Probe (Ping): Zero-overhead health check returning server status (UP), connector version, and JVM system load average.
  • Intelligent Claim-Check Pattern Offloading:
    • Inlined binary payloads (raw_bytes) or large text payloads (text_content) smaller than inlineBinaryThresholdKb (default: 512 KB) flow through memory without extra disk round-trips before being persisted by JobOrchestrator.
    • Payloads exceeding inlineBinaryThresholdKb are automatically streamed into the active ClaimCheckStore (Local filesystem, Apache Ozone ofs/s3g, or AWS S3) by OisDocumentTranslator, replacing the in-memory byte array with a lightweight claimCheckUri reference.
    • Direct Pre-Uploaded Claim-Check Support: Producers handling multi-gigabyte archives or media files can upload directly to S3 or Apache Ozone first and pass the resulting claim_check_uri in the Protobuf payload—completely bypassing gRPC payload transfer while still triggering OpenCrawling's Tika extraction and RAG pipeline.
  • Tombstone DELETE Short-Circuiting:
    • Pushing a document with action = DELETE creates a tombstone RepositoryDocument (RepositoryDocument.createTombstone(docId)).
    • JobOrchestrator immediately short-circuits Apache Tika text extraction, chunking, and embedding generation, publishing a deletion event that purges all associated vector chunks from downstream vector databases (pgvector, Milvus, Qdrant, Vespa, OpenSearch, Solr 10).
  • Zero-Trust Security & Multi-Tenant ACL Mapping:
    • Translates Protobuf SecurityConfig (allow_sids, deny_sids, owner, tenant_id) and legacy acl lists into OpenCrawling's unified org.opencrawling.core.security.SecurityConfig (PermissionRule with read and deny access levels) and flat ACL filter strings.
    • Preserves explicit deny_sids precedence and stamps tenant_id into document metadata for strict multi-tenant vector isolation.
  • Enterprise Transport Security (Bearer Token, TLS & mTLS):
    • Bearer Token Authentication: GrpcSecurityInterceptor validates Authorization: Bearer <token> metadata headers using constant-time comparison while exempting the diagnostic Ping RPC for liveness probes.
    • TLS & Mutual TLS (mTLS): GrpcPushServerManager configures Netty SslContextBuilder with X.509 server certificates, PKCS#8 private keys, custom CA trust stores, and configurable ClientAuth (NONE, OPTIONAL, REQUIRE).
  • End-to-End W3C Distributed Tracing:
    • OpenTelemetryServerInterceptor extracts W3C Trace Context headers (traceparent and tracestate) from incoming gRPC metadata via W3CTraceContextPropagator, linking upstream producer spans directly with OpenCrawling's Kafka, Tika, Embedding, and Vector Writer spans.
  • Java 25 Virtual Thread Execution:
    • The gRPC Netty server dispatches every incoming RPC onto a dedicated Java 25 Virtual Thread (Executors.newVirtualThreadPerTaskExecutor()), enabling thousands of concurrent streaming connections with minimal memory footprint.
  • Lakehouse Audit Reporting & Document Ledger Integration:
    • Every unary PushDocument call (exec-push-<8char>) and streaming StreamDocuments / BiDirectionalStream session (exec-stream-<8char>) is automatically tracked in JobLedgerAccumulator (UPSERT as SUCCESS/FAILED, DELETE as TOMBSTONED), updates live job execution counters in JobController, and synchronously writes Hive-partitioned Apache Parquet, CSV Compliance Manifest, and Operational JSON Summary reports queryable in the Admin UI (Reports & Audit Ledger) and DuckDB SQL Playground.

📜 Protobuf Contract (ois_push_service.proto)

The Protobuf contract is located in oc-grpc-push-connector/src/main/proto/opencrawling/push/v1/ois_push_service.proto:

syntax = "proto3";

package opencrawling.push.v1;

option java_multiple_files = true;
option java_package = "org.opencrawling.push.v1";
option java_outer_classname = "OisPushServiceProto";

service OisPushService {
  // 1. Unary: Push a single OIS document (for low-volume or transactional event hooks)
  rpc PushDocument (PushDocumentRequest) returns (PushDocumentResponse);

  // 2. Client Streaming: Push a high-throughput batch/stream of documents
  rpc StreamDocuments (stream PushDocumentRequest) returns (StreamSummaryResponse);

  // 3. Bi-Directional Streaming: Real-time reactive ingestion with per-document ACK/NACK and backpressure
  rpc BiDirectionalStream (stream PushDocumentRequest) returns (stream IngestionAck);

  // 4. Health & Diagnostic Ping (Used by Admin UI & CLI to verify connectivity)
  rpc Ping (PingRequest) returns (PingResponse);
}

enum DocumentAction {
  DOCUMENT_ACTION_UNSPECIFIED = 0;
  UPSERT = 1;
  DELETE = 2;
}

enum IngestionStatus {
  INGESTION_STATUS_UNSPECIFIED = 0;
  ACCEPTED = 1;
  OFFLOADED_TO_CLAIM_CHECK = 2;
  REJECTED_INVALID_PAYLOAD = 3;
  REJECTED_UNAUTHORIZED = 4;
  INTERNAL_ERROR = 5;
}

message SecurityConfig {
  repeated string allow_sids = 1;
  repeated string deny_sids = 2;
  string owner = 3;
  string tenant_id = 4;
}

message OisDocumentPayload {
  string document_id = 1;                     // Unique URI or repository ID (required)
  string title = 2;                           // Human-readable title
  string mime_type = 3;                       // e.g., "application/pdf", "text/markdown"
  DocumentAction action = 4;                  // UPSERT or DELETE (Tombstone)

  oneof content_source {
    bytes raw_bytes = 5;                      // Inline binary content (up to threshold, e.g., 4MB-16MB)
    string text_content = 6;                  // Pre-extracted plain text or Markdown
    string claim_check_uri = 7;               // Pre-uploaded Ozone/S3 URI (for massive multi-GB files)
  }

  map<string, string> metadata = 8;           // Arbitrary key-value metadata
  SecurityConfig security = 9;                // Zero-Trust ACLs
  repeated string acl = 10;                   // Flat ACL list for backward compatibility
  int64 created_at_epoch_ms = 11;
  int64 updated_at_epoch_ms = 12;
}

message PushOptions {
  string job_id = 1;                          // Target OpenCrawling Job ID
  bool sync_mode = 2;                         // If true, wait for Kafka/pipeline ACK before returning
  bool extract_archive = 3;                   // Hint for Tika / Archive expansion
  string routing_tag = 4;                     // Optional routing tag for custom transformation pipelines
}

message PushDocumentRequest {
  string correlation_id = 1;                  // Trace / correlation ID
  PushOptions options = 2;
  OisDocumentPayload document = 3;
}

message PushDocumentResponse {
  string correlation_id = 1;
  string document_id = 2;
  IngestionStatus status = 3;
  string claim_check_uri = 4;                 // Populated if binary was offloaded to Ozone/S3
  string error_message = 5;
  int64 processing_time_ms = 6;
}

message StreamSummaryResponse {
  int64 total_received = 1;
  int64 accepted_count = 2;
  int64 rejected_count = 3;
  repeated PushFailure failures = 4;
  int64 total_time_ms = 5;
}

message PushFailure {
  string correlation_id = 1;
  string document_id = 2;
  IngestionStatus status = 3;
  string error_message = 4;
}

message IngestionAck {
  string correlation_id = 1;
  string document_id = 2;
  IngestionStatus status = 3;
  string message = 4;
  int64 server_timestamp = 5;
}

message PingRequest {
  string client_id = 1;
}

message PingResponse {
  string status = 1;                          // "UP"
  string version = 2;
  int64 active_streams = 3;
  double system_load = 4;
}

⚙️ Configuration Reference

1. Spring Boot Properties (application.yml)

opencrawling:
  connector:
    grpc-push:
      enabled: ${OPENCRAWLING_GRPC_PUSH_ENABLED:true}
      port: ${OPENCRAWLING_GRPC_PUSH_PORT:9096}
      max-message-size-mb: ${OPENCRAWLING_GRPC_PUSH_MAX_MESSAGE_SIZE_MB:64}
      inline-binary-threshold-kb: ${OPENCRAWLING_GRPC_PUSH_INLINE_THRESHOLD_KB:512}
      token: ${OPENCRAWLING_GRPC_PUSH_TOKEN:}
      reflection-enabled: ${OPENCRAWLING_GRPC_PUSH_REFLECTION_ENABLED:true}
      tls:
        enabled: ${OPENCRAWLING_GRPC_PUSH_TLS_ENABLED:false}
        cert-chain-path: ${OPENCRAWLING_GRPC_PUSH_TLS_CERT:}
        private-key-path: ${OPENCRAWLING_GRPC_PUSH_TLS_KEY:}
        trust-cert-collection-path: ${OPENCRAWLING_GRPC_PUSH_TLS_TRUST_CERT:}
        client-auth: ${OPENCRAWLING_GRPC_PUSH_TLS_CLIENT_AUTH:NONE}

2. Environment Variables

Environment Variable Property Key Default Description
OPENCRAWLING_GRPC_PUSH_ENABLED opencrawling.connector.grpc-push.enabled true Enable or disable the embedded gRPC Push Server
OPENCRAWLING_GRPC_PUSH_PORT opencrawling.connector.grpc-push.port 9096 TCP port on which the Netty gRPC Push server listens
OPENCRAWLING_GRPC_PUSH_MAX_MESSAGE_SIZE_MB opencrawling.connector.grpc-push.max-message-size-mb 64 Maximum inbound gRPC message size in megabytes
OPENCRAWLING_GRPC_PUSH_INLINE_THRESHOLD_KB opencrawling.connector.grpc-push.inline-binary-threshold-kb 512 Inline payload size threshold in KB before automatic Claim-Check offload
OPENCRAWLING_GRPC_PUSH_TOKEN opencrawling.connector.grpc-push.token (empty) Optional static Bearer token required in Authorization: Bearer <token> metadata
OPENCRAWLING_GRPC_PUSH_REFLECTION_ENABLED opencrawling.connector.grpc-push.reflection-enabled true Enables gRPC Server Reflection (ProtoReflectionService) for grpcurl / Postman
OPENCRAWLING_GRPC_PUSH_TLS_ENABLED opencrawling.connector.grpc-push.tls.enabled false Enables TLS encryption on the gRPC Push listener
OPENCRAWLING_GRPC_PUSH_TLS_CERT opencrawling.connector.grpc-push.tls.cert-chain-path (empty) File path to server X.509 certificate chain (PEM)
OPENCRAWLING_GRPC_PUSH_TLS_KEY opencrawling.connector.grpc-push.tls.private-key-path (empty) File path to server private key (PKCS#8 PEM)
OPENCRAWLING_GRPC_PUSH_TLS_TRUST_CERT opencrawling.connector.grpc-push.tls.trust-cert-collection-path (empty) File path to trusted CA certificates for mTLS client verification
OPENCRAWLING_GRPC_PUSH_TLS_CLIENT_AUTH opencrawling.connector.grpc-push.tls.client-auth NONE Mutual TLS client certificate mode: NONE, OPTIONAL, or REQUIRE

🖥️ Admin UI Configuration & Live Diagnostic Probe

In the OpenCrawling Admin Dashboard (oc-admin-ui):

  1. Navigate to Connectors -> New Connector.
  2. Select Repository (Source) as the Connector Type.
  3. Choose gRPC Push Streaming Ingestion (org.opencrawling.push.grpc.GrpcPushRepositoryConnector).
  4. Configure the gRPC Push Endpoint Configuration panel:
    • Listener Port: 9096
    • Max Message Size (MB): 64
    • Inline Binary Threshold (KB): 512
    • Authentication Token (Bearer): Optional secret token
    • Mutual TLS Client Auth: NONE, OPTIONAL, or REQUIRE
    • Enable TLS Encryption: Toggle TLS / mTLS certificate paths
  5. Click Test gRPC Push Listener (Ping) to trigger a live gRPC Ping RPC via ConnectorCheckerService (POST /api/connectors/check) and verify the listener's status and version.

🛠️ Usage Examples

0. OpenCrawling CLI (oc push)

The OpenCrawling CLI (oc-cli) provides native oc push subcommands (ping, upsert, delete, stream) and oc connector check overrides:

# Probe gRPC Push listener health and round-trip latency
oc push ping --host 127.0.0.1 --port 9096

# Verify via Admin REST API Connector Checker
oc connector check --name Grpc_Push_Connector --type repository --config host=127.0.0.1 --config port=9096

# Push an inline or file-based UPSERT document with Zero-Trust ACLs
oc push upsert \
  --job-id 1 \
  --id "https://ecm.example.com/contracts/2026-msa.md" \
  --text "# Master Services Agreement..." \
  --acl-allow "GROUP_LEGAL,ROLE_EXECUTIVE" \
  --acl-deny "GROUP_CONTRACTORS" \
  --metadata "department=legal" \
  --token "my-secret-token"

# Stream a local directory of files over Bi-Directional HTTP/2 gRPC
oc push stream --bidi --job-id 1 --dir ./docs/contracts --token "my-secret-token"

# Push a DELETE tombstone
oc push delete --job-id 1 --id "https://ecm.example.com/contracts/2026-msa.md" --token "my-secret-token"

1. Diagnostic Health Check with grpcurl

Because gRPC Server Reflection is enabled by default, you can inspect and query the service directly with grpcurl:

# List exposed gRPC services
grpcurl -plaintext 127.0.0.1:9096 list

# Execute diagnostic Ping RPC
grpcurl -plaintext \
  -d '{"client_id": "devops-cli"}' \
  127.0.0.1:9096 opencrawling.push.v1.OisPushService/Ping

Response:

{
  "status": "UP",
  "version": "1.0.0-SNAPSHOT",
  "systemLoad": 0.14
}

2. Unary PushDocument (UPSERT) with Zero-Trust ACLs

grpcurl -plaintext \
  -H "authorization: Bearer my-secret-token" \
  -d '{
    "correlation_id": "req-1001",
    "options": {
      "job_id": "1",
      "sync_mode": true,
      "routing_tag": "legal-contracts"
    },
    "document": {
      "document_id": "https://ecm.example.com/contracts/2026-msa.md",
      "title": "Master Services Agreement 2026",
      "mime_type": "text/markdown",
      "action": "UPSERT",
      "text_content": "# Master Services Agreement\n\nThis agreement governs enterprise cloud services...",
      "metadata": {
        "department": "legal",
        "contract_year": "2026"
      },
      "security": {
        "allow_sids": ["GROUP_LEGAL", "ROLE_EXECUTIVE"],
        "deny_sids": ["GROUP_CONTRACTORS"],
        "owner": "alice.smith",
        "tenant_id": "acme-corp"
      }
    }
  }' \
  127.0.0.1:9096 opencrawling.push.v1.OisPushService/PushDocument

Response:

{
  "correlationId": "req-1001",
  "documentId": "https://ecm.example.com/contracts/2026-msa.md",
  "status": "ACCEPTED",
  "processingTimeMs": "4"
}

3. Pushing a DELETE Tombstone

When a record is deleted in the source system, push a DELETE tombstone to immediately purge its vectors across all configured vector stores:

grpcurl -plaintext \
  -H "authorization: Bearer my-secret-token" \
  -d '{
    "correlation_id": "req-1002",
    "options": {
      "job_id": "1"
    },
    "document": {
      "document_id": "https://ecm.example.com/contracts/2026-msa.md",
      "action": "DELETE"
    }
  }' \
  127.0.0.1:9096 opencrawling.push.v1.OisPushService/PushDocument

4. Java 25 Producer Client (Unary & Bi-Directional Streaming)

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.Metadata;
import io.grpc.stub.MetadataUtils;
import io.grpc.stub.StreamObserver;
import org.opencrawling.push.v1.*;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class OpenCrawlingGrpcPushClientExample {

    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 9096)
                .usePlaintext()
                .build();

        Metadata headers = new Metadata();
        headers.put(Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER), "Bearer my-secret-token");

        // 1. Unary Push
        OisPushServiceGrpc.OisPushServiceBlockingStub blockingStub =
                OisPushServiceGrpc.newBlockingStub(channel)
                        .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers));

        PushDocumentResponse unaryResp = blockingStub.pushDocument(PushDocumentRequest.newBuilder()
                .setCorrelationId("unary-1")
                .setOptions(PushOptions.newBuilder().setJobId("1").setSyncMode(true).build())
                .setDocument(OisDocumentPayload.newBuilder()
                        .setDocumentId("doc://kb/architecture-overview")
                        .setTitle("Architecture Overview")
                        .setMimeType("text/plain")
                        .setAction(DocumentAction.UPSERT)
                        .setTextContent("OpenCrawling decouples ingestion via Kafka and Claim-Check storage.")
                        .setSecurity(SecurityConfig.newBuilder()
                                .addAllowSids("GROUP_ENGINEERING")
                                .setOwner("admin")
                                .setTenantId("tenant-1")
                                .build())
                        .build())
                .build());

        System.out.println("Unary Status: " + unaryResp.getStatus());

        // 2. Bi-Directional Streaming Push
        OisPushServiceGrpc.OisPushServiceStub asyncStub =
                OisPushServiceGrpc.newStub(channel)
                        .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers));

        CountDownLatch done = new CountDownLatch(1);
        StreamObserver<PushDocumentRequest> requestObserver = asyncStub.biDirectionalStream(
                new StreamObserver<>() {
                    @Override
                    public void onNext(IngestionAck ack) {
                        System.out.printf("ACK [%s] doc=%s status=%s%n",
                                ack.getCorrelationId(), ack.getDocumentId(), ack.getStatus());
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                        done.countDown();
                    }

                    @Override
                    public void onCompleted() {
                        done.countDown();
                    }
                });

        for (int i = 1; i <= 5; i++) {
            requestObserver.onNext(PushDocumentRequest.newBuilder()
                    .setCorrelationId("stream-" + i)
                    .setOptions(PushOptions.newBuilder().setJobId("1").build())
                    .setDocument(OisDocumentPayload.newBuilder()
                            .setDocumentId("doc://kb/item-" + i)
                            .setAction(DocumentAction.UPSERT)
                            .setMimeType("text/plain")
                            .setTextContent("Streaming document payload #" + i)
                            .build())
                    .build());
        }
        requestObserver.onCompleted();
        done.await(10, TimeUnit.SECONDS);
        channel.shutdownNow();
    }
}

🧪 Testing & Verification

Run the automated test suites via the integration test runner or directly:

# Run both the unit/in-process suite and the full Docker decoupled pipeline test
./run-integration-tests.sh grpc-push

# Or run each suite individually:
./scripts/test-grpc-push-connector.sh
./scripts/test-grpc-push-decoupled.sh

The test suites execute:

  1. GrpcPushConnectorTest: Verifies OisDocumentTranslator inline vs. Claim-Check offloading thresholds, pre-uploaded claim_check_uri pass-through, Zero-Trust SecurityConfig and deny_sids translation, DELETE tombstone creation, GrpcPushRepositoryConnector schema introspection, and GrpcPushClientCli end-to-end TCP execution.
  2. OisPushServiceInProcessTest: Full in-process gRPC server/client verification covering Unary PushDocument, Client Streaming StreamDocuments, Bi-Directional Streaming BiDirectionalStream, Ping, Bearer token enforcement (GrpcSecurityInterceptor), and W3C traceparent propagation (OpenTelemetryServerInterceptor).
  3. RuntimePushDocumentDispatcherTest: Verifies oc-runtime integration with JobController, JobOrchestrator, and JobLedgerAccumulator across RAG and Migration pipeline modes, including synchronous generation of Parquet, CSV, and JSON audit reports for unary (exec-push-*) and streaming (exec-stream-*) push executions.
  4. test-grpc-push-decoupled.sh: Spins up the full decoupled Docker stack (docker-compose-decoupled.yml), verifies Ping and Admin REST API /api/connectors/check, asserts UNAUTHENTICATED rejection without a Bearer token, pushes an authenticated UPSERT over gRPC port 9096 through the decoupled Kafka + Claim-Check + Ollama + PGVector consumer pipeline, verifies that an exec-push-* execution report is generated in /api/reports, and verifies vector removal via a gRPC DELETE tombstone.

🗺️ Documentation

🧪 Testing

🤝 Community

Clone this wiki locally