Implementation:Datahub project Datahub S3Emitter

Knowledge Sources	Datahub_project_Datahub
Domains	Java_SDK, Metadata_Emission, AWS_S3
Last Updated	2026-02-10 00:00 GMT

Overview

An Emitter implementation that collects metadata change proposals into a local temporary file and uploads them to an AWS S3 bucket upon close.

Description

S3Emitter implements the Emitter interface to provide a two-phase metadata emission strategy: metadata changes are first written to a local temporary JSON file via a delegated FileEmitter, and then uploaded to S3 when the emitter is closed.

Construction: The constructor creates a temporary file (prefixed datahub_ingest_ with suffix _mcps.json), initializes a FileEmitter to write to it, and builds an AWS S3Client based on the provided S3EmitterConfig. The S3 client supports:

• Static credentials -- If accessKey and secretKey are provided, uses StaticCredentialsProvider.
• Default credentials chain -- Otherwise, uses DefaultCredentialsProvider with optional profile name and profile file.
• Custom endpoint -- For S3-compatible services (e.g., MinIO).
• Custom region -- AWS region override.

Emission: Both emit(MetadataChangeProposalWrapper, Callback) and emit(MetadataChangeProposal, Callback) delegate directly to the underlying FileEmitter, accumulating MCPs in the temporary file. UpsertAspectRequest emission is not supported and throws UnsupportedOperationException.

Close and Upload: On close(), the file emitter is flushed and closed, the S3 object key is computed from the config's pathPrefix and either the configured fileName or the temporary file name, and the file is uploaded via PutObjectRequest. The temporary file is deleted after upload. An IOException is thrown if the upload fails.

Connection Testing: Not supported; throws UnsupportedOperationException.

Usage

Use S3Emitter when metadata changes should be written to S3 for later processing or archival, rather than sent directly to the DataHub GMS REST API or Kafka. This is useful for offline ingestion, batch processing pipelines, or environments where direct API access is not available.

Code Reference

Source Location

metadata-integration/java/datahub-client/src/main/java/datahub/client/s3/S3Emitter.java

Signature

public class S3Emitter implements Emitter {

    public S3Emitter(S3EmitterConfig config) throws IOException

    public Future<MetadataWriteResponse> emit(
        MetadataChangeProposalWrapper mcpw, Callback callback) throws IOException
    public Future<MetadataWriteResponse> emit(
        MetadataChangeProposal mcp, Callback callback) throws IOException
    public Future<MetadataWriteResponse> emit(
        List<UpsertAspectRequest> request, Callback callback) throws IOException

    public boolean testConnection()
        throws IOException, ExecutionException, InterruptedException
    public void close() throws IOException
}

Import

import datahub.client.s3.S3Emitter;
import datahub.client.s3.S3EmitterConfig;

I/O Contract

Inputs

Method	Parameter	Type	Description
Constructor	config	S3EmitterConfig	S3 bucket, credentials, and path settings
emit	mcpw	MetadataChangeProposalWrapper	High-level metadata change proposal
emit	mcp	MetadataChangeProposal	Raw metadata change proposal
emit	callback	Callback	Asynchronous completion callback

Outputs

• emit() returns Future<MetadataWriteResponse> from the underlying FileEmitter.
• close() uploads the accumulated JSON file to S3 at s3://{bucketName}/{pathPrefix}/{fileName}. Throws IOException on upload failure.
• The temporary local file is automatically cleaned up after upload.

Usage Examples

S3EmitterConfig config = S3EmitterConfig.builder()
    .bucketName("my-datahub-bucket")
    .pathPrefix("ingestion/metadata")
    .region("us-east-1")
    .build();

try (S3Emitter emitter = new S3Emitter(config)) {
    // Emit multiple MCPs (accumulated in temp file)
    emitter.emit(mcpWrapper1, callback);
    emitter.emit(mcpWrapper2, callback);
    // close() triggers upload to S3
}

// With static credentials and custom endpoint (e.g., MinIO)
S3EmitterConfig minioConfig = S3EmitterConfig.builder()
    .bucketName("datahub")
    .pathPrefix("mcps")
    .endpoint("http://minio:9000")
    .accessKey("minioadmin")
    .secretKey("minioadmin")
    .region("us-east-1")
    .build();

Related Pages

• Datahub_project_Datahub_S3EmitterConfig -- Configuration for S3Emitter
• Datahub_project_Datahub_MetadataWriteCallback_Interface -- Callback interface for async notification
• Datahub_project_Datahub_KafkaEmitter -- Alternative emitter targeting Kafka
• Datahub_project_Datahub_RestEmitterConfig -- Configuration for the REST-based emitter