Implementation:Datahub project Datahub RestEmitter Create
| Property | Value |
|---|---|
| Implementation Name | RestEmitter_Create |
| Type | API Doc |
| Category | Java_SDK_Metadata_Emission |
| Workflow | Java_SDK_Metadata_Emission |
| Repository | https://github.com/datahub-project/datahub |
| Last Updated | 2026-02-09 17:00 GMT |
Overview
Description
RestEmitter_Create documents the factory methods and constructors for creating DataHub emitter instances. The primary entry point is RestEmitter.create(), a static factory method that accepts a lambda-based builder configuration and returns a fully initialized RestEmitter connected to a DataHub GMS server over HTTP. The class also provides RestEmitter.createWithDefaults() for zero-configuration instantiation targeting http://localhost:8080.
In addition to RestEmitter, the SDK provides three other emitter implementations that share the Emitter interface: KafkaEmitter (constructed via new KafkaEmitter(KafkaEmitterConfig)), FileEmitter (constructed via new FileEmitter(FileEmitterConfig)), and S3Emitter (constructed via new S3Emitter(S3EmitterConfig)).
Usage
Call RestEmitter.create() with a lambda that configures the server URL, authentication token, timeouts, and other connection properties. The returned emitter is thread-safe and ready to accept emit() calls immediately. For non-REST backends, construct the appropriate emitter class directly with its config object.
Code Reference
Source Location
| Property | Value |
|---|---|
| File | metadata-integration/java/datahub-client/src/main/java/datahub/client/rest/RestEmitter.java
|
| Lines | L72-422 |
| Repository | https://github.com/datahub-project/datahub |
Signature
RestEmitter factory method:
public static RestEmitter create(
Consumer<RestEmitterConfig.RestEmitterConfigBuilder> builderSupplier
) -> RestEmitter
RestEmitter default factory:
public static RestEmitter createWithDefaults() -> RestEmitter
RestEmitter direct constructor:
public RestEmitter(RestEmitterConfig config)
KafkaEmitter constructor:
public KafkaEmitter(KafkaEmitterConfig config) throws IOException
public KafkaEmitter(KafkaEmitterConfig config, String mcpKafkaTopic) throws IOException
FileEmitter constructor:
public FileEmitter(FileEmitterConfig config)
S3Emitter constructor:
public S3Emitter(S3EmitterConfig config) throws IOException
Import
import datahub.client.rest.RestEmitter;
import datahub.client.rest.RestEmitterConfig;
import datahub.client.kafka.KafkaEmitter;
import datahub.client.kafka.KafkaEmitterConfig;
import datahub.client.file.FileEmitter;
import datahub.client.file.FileEmitterConfig;
import datahub.client.s3.S3Emitter;
import datahub.client.s3.S3EmitterConfig;
RestEmitter Configuration Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| server | String | "http://localhost:8080" |
URL of the DataHub GMS server |
| token | String | null |
Bearer authentication token for DataHub Cloud or authenticated instances |
| timeoutSec | Integer | null (uses HttpClient defaults of 10s connect / 10s read) |
Overrides both connection request timeout and response timeout in seconds |
| disableSslVerification | boolean | false |
When true, disables SSL certificate and hostname verification (insecure)
|
| disableChunkedEncoding | boolean | false |
When true, sends request body as a byte array instead of chunked transfer encoding
|
| maxRetries | int | 0 |
Maximum number of retry attempts for failed HTTP requests |
| retryIntervalSec | int | 10 |
Interval in seconds between retry attempts |
| extraHeaders | Map<String, String> | empty map | Additional HTTP headers to include in every request |
| eventFormatter | EventFormatter | EventFormatter(PEGASUS_JSON) |
Controls serialization format for metadata events |
| asyncIngest | Boolean | null |
When set, passes async parameter to the ingest endpoint
|
I/O Contract
| Direction | Description |
|---|---|
| Input | Configuration parameters: server URL, authentication token, timeout, SSL settings, retry policy, extra headers |
| Output | A fully initialized RestEmitter (or other Emitter implementation) with an active HTTP async client (or Kafka producer, or file writer) ready to accept emit() calls
|
| Side Effects | RestEmitter: starts an Apache CloseableHttpAsyncClient with connection pool. KafkaEmitter: creates a KafkaProducer. FileEmitter: opens a BufferedWriter and writes opening [ bracket. S3Emitter: creates a temporary file and initializes an S3 client.
|
Usage Examples
Example 1: RestEmitter with custom configuration
import datahub.client.rest.RestEmitter;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
RestEmitter emitter = RestEmitter.create(b -> b
.server("http://localhost:8080")
.token("your-auth-token")
.timeoutSec(30)
.extraHeaders(Collections.singletonMap("Session-token", "MY_SESSION"))
.customizeHttpAsyncClient(c -> c.setConnectionTimeToLive(30, TimeUnit.SECONDS))
);
Example 2: RestEmitter with defaults
import datahub.client.rest.RestEmitter;
RestEmitter emitter = RestEmitter.createWithDefaults();
Example 3: KafkaEmitter
import datahub.client.kafka.KafkaEmitter;
import datahub.client.kafka.KafkaEmitterConfig;
KafkaEmitterConfig config = KafkaEmitterConfig.builder().build();
KafkaEmitter emitter = new KafkaEmitter(config);
Example 4: FileEmitter
import datahub.client.file.FileEmitter;
import datahub.client.file.FileEmitterConfig;
FileEmitter emitter = new FileEmitter(
FileEmitterConfig.builder()
.fileName("/my/path/output.json")
.build()
);
Example 5: S3Emitter
import datahub.client.s3.S3Emitter;
import datahub.client.s3.S3EmitterConfig;
S3Emitter emitter = new S3Emitter(
S3EmitterConfig.builder()
.bucketName("my-metadata-bucket")
.pathPrefix("datahub/mcps")
.region("us-east-1")
.build()
);