Bigtable mengubah aliran data ke template Vector Search

Template ini membuat pipeline streaming untuk melakukan streaming kumpulan data perubahan data Bigtable dan menulisnya ke Vertex AI Vector Search menggunakan Dataflow Runner V2.

Persyaratan pipeline

• Instance sumber Bigtable harus ada.
• Tabel sumber Bigtable harus ada, dan tabel harus mengaktifkan aliran data perubahan.
• Profil aplikasi Bigtable harus ada.
• Jalur indeks Penelusuran Vektor harus ada.

Parameter template

Parameter	Deskripsi
embeddingColumn	Nama kolom yang sepenuhnya memenuhi syarat tempat embeddings disimpan. Dalam format cf:col.
embeddingByteSize	Ukuran byte setiap entri dalam array embedding. Gunakan 4 untuk {i>float<i} dan 8 untuk {i>double<i}. Nilai defaultnya adalah 4.
vectorSearchIndex	Indeks Vector Search tempat perubahan akan di-streaming, dalam format &#39;projects/{projectID}/locations/{region}/indexes/{indexID}&#39; (tidak boleh ada spasi di bagian awal atau akhir). Contoh: projects/123/locations/us-east1/indexes/456.
bigtableChangeStreamAppProfile	Profil aplikasi yang digunakan untuk membedakan beban kerja di Bigtable.
bigtableReadInstanceId	ID instance Bigtable yang berisi tabel.
bigtableReadTableId	Tabel Bigtable untuk dibaca.
bigtableMetadataTableTableId	Opsional: ID untuk tabel metadata yang dibuat. Jika tidak disetel, Bigtable akan menghasilkan ID.
crowdingTagColumn	Opsional: Nama kolom yang sepenuhnya memenuhi syarat tempat tag kerumunan disimpan, dalam format cf:col.
allowRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai pembatasan allow, plus alias mereka. Setiap nama kolom harus menggunakan format cf:col->alias.
denyRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma dari kolom yang akan digunakan sebagai pembatasan deny, plus alias mereka. Setiap nama kolom harus menggunakan format cf:col->alias.
intNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma yang digunakan sebagai bilangan bulat numeric_restricts, plus alias mereka. Setiap nama kolom harus menggunakan format cf:col->alias.
floatNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma untuk digunakan sebagai float (4 byte) numeric_restricts, plus alias mereka. Setiap nama kolom harus dalam format cf:col->alias
doubleNumericRestrictsMappings	Opsional: Nama kolom yang sepenuhnya memenuhi syarat dan dipisahkan koma untuk digunakan sebagai kolom ganda (8 byte) numeric_restricts, plus alias mereka. Setiap nama kolom harus dalam format cf:col->alias
upsertMaxBatchSize	Opsional: Jumlah maksimum pembaruan dan penyisipan yang akan di-buffer sebelum pembaruan dan penyisipan batch ke indeks Vector Search. Batch dikirim saat ada upsertBatchSize catatan yang siap. Contoh: 10.
upsertMaxBufferDuration	Opsional: Penundaan maksimum sebelum sekumpulan pembaruan dan penyisipan dikirim ke Vector Search. {i>Batch<i} dikirim ketika ada upsertBatchSize data siap. Format yang diizinkan adalah: Ns untuk detik (contoh: 5 dtk), Nm untuk menit (contoh: 12 mnt), dan Nh untuk jam (contoh: 2 jam). Default: 10s.
deleteMaxBatchSize	Opsional: Jumlah maksimum penghapusan yang akan di-buffer sebelum menghapus batch dari indeks Vector Search. {i>Batch<i} dikirim ketika ada deleteBatchSize data siap. Contoh: 10.
deleteMaxBufferDuration	Opsional: Penundaan maksimum sebelum batch penghapusan dikirim ke Penelusuran Vektor. {i>Batch<i} dikirim ketika ada salah satu data deleteBatchSize siap. Format yang diizinkan adalah: Ns untuk detik (contoh: 5 dtk), Nm untuk menit (contoh: 12 mnt), dan Nh untuk jam (contoh: 2 jam). Default: 10s.
dlqDirectory	Opsional: Jalur untuk menyimpan data yang belum diproses dengan alasan kegagalan tersebut untuk diproses. Defaultnya adalah direktori yang berada di lokasi sementara tugas Dataflow. Nilai default sesuai untuk sebagian besar skenario.
bigtableChangeStreamMetadataInstanceId	Opsional: Instance Bigtable yang akan digunakan untuk tabel metadata konektor aliran perubahan. Nilai defaultnya adalah kosong.
bigtableChangeStreamMetadataTableTableId	Opsional: ID tabel metadata konektor perubahan stream Bigtable yang akan digunakan. Jika tidak diberikan, Tabel metadata konektor aliran perubahan Bigtable otomatis dibuat selama pipeline alur kerja. Nilai defaultnya adalah kosong.
bigtableChangeStreamCharset	Opsional: Bigtable mengubah aliran nama charset saat membaca nilai dan penentu kolom. Defaultnya adalah UTF-8.
bigtableChangeStreamStartTimestamp	Opsional: DateTime awal, inklusif, yang akan digunakan untuk membaca aliran data perubahan (https://tools.ietf.org/html/rfc3339). Misalnya, 2022-05-05T07:59:59Z. Setelan defaultnya adalah stempel waktu saat pipeline dimulai.
bigtableChangeStreamIgnoreColumnFamilies	Opsional: Daftar perubahan nama keluarga kolom yang dipisahkan koma yang tidak akan ditangkap. Nilai defaultnya adalah kosong.
bigtableChangeStreamIgnoreColumns	Opsional: Daftar perubahan nama kolom yang dipisahkan koma yang tidak akan ditangkap. Nilai defaultnya adalah kosong.
bigtableChangeStreamName	Opsional: Nama unik untuk pipeline klien. Parameter ini Ini berarti Anda melanjutkan pemrosesan dari titik di mana pipeline yang berjalan sebelumnya terhenti. Setelan defaultnya adalah nama yang dibuat secara otomatis. Lihat log tugas Dataflow untuk nilai yang digunakan.
bigtableChangeStreamResume	Opsional: Jika disetel ke benar (true), pipeline baru akan dilanjutkan dari titik di mana operasi yang sebelumnya berjalan pipeline dengan nama yang sama telah dihentikan. Jika sebuah pipeline dengan belum pernah berjalan, {i>pipelines<i} yang baru akan gagal dimulai. Gunakan parameter bigtableChangeStreamName untuk menentukan jalur pipeline. Jika disetel ke salah (false), pipeline baru akan dimulai. Jika suatu pipeline dengan nama yang sama dengan bigtableChangeStreamName sudah berjalan di masa lalu untuk sumber tertentu, pipeline baru gagal dimulai. Nilai defaultnya adalah false (salah).
bigtableReadProjectId	Opsional: Project yang akan dibaca data Bigtable. Default untuk parameter ini adalah project tempat Pipeline Dataflow sedang berjalan.

Menjalankan template

Konsol

1. Buka halaman Create job from template Dataflow.
Buka Buat tugas dari template
2. Di kolom Job name, masukkan nama tugas yang unik.
3. Opsional: Untuk Regional endpoint, pilih nilai dari menu drop-down. Default region adalah us-central1.

   Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Dataflow template, pilih the Bigtable Change Streams to Vector Search template.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Run job.

gcloud CLI

Catatan: Untuk menggunakan Google Cloud CLI untuk menjalankan flex template, Anda harus memiliki versi Google Cloud CLI 284.0.0 atau yang lebih baru.

Di shell atau terminal Anda, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_Vector_Search \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       embeddingColumn=EMBEDDING_COLUMN,\
       embeddingByteSize=EMBEDDING_BYTE_SIZE,\
       vectorSearchIndex=VECTOR_SEARCH_INDEX,\
       bigtableChangeStreamAppProfile=BIGTABLE_CHANGE_STREAM_APP_PROFILE,\
       bigtableReadInstanceId=BIGTABLE_READ_INSTANCE_ID,\
       bigtableReadTableId=BIGTABLE_READ_TABLE_ID,\

Ganti kode berikut:

• JOB_NAME: nama pekerjaan unik pilihan Anda
• VERSION: versi {i>template<i} yang ingin Anda gunakan.

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di Folder induk non-dated di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi tertentu dari , yang dapat ditemukan disarangkan di masing-masing folder induk bertanggal dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi template terbaru mungkin diupdate dengan menyebabkan gangguan perubahan. Lingkungan produksi Anda harus menggunakan template yang disimpan dalam versi terbaru dated untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi di alur kerja produksi.
• REGION_NAME: wilayah tempat Anda ingin men-deploy tugas Dataflow Anda—misalnya, us-central1
• EMBEDDING_COLUMN: kolom Penyematan
• EMBEDDING_BYTE_SIZE: Ukuran byte dari array embedding. Bisa 4 atau 8.
• VECTOR_SEARCH_INDEX: Jalur indeks Penelusuran Vektor
• BIGTABLE_CHANGE_STREAM_APP_PROFILE: ID profil aplikasi Bigtable
• BIGTABLE_READ_INSTANCE_ID: ID Instance Bigtable sumber
• BIGTABLE_READ_TABLE_ID: ID tabel Bigtable sumber

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk informasi selengkapnya tentang API dan cakupan otorisasinya, lihat projects.templates.launch

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launchParameter": {
     "jobName": "JOB_NAME",
     "parameters": {
       "embeddingColumn": "EMBEDDING_COLUMN",
       "embeddingByteSize": "EMBEDDING_BYTE_SIZE",
       "vectorSearchIndex": "VECTOR_SEARCH_INDEX",
       "bigtableChangeStreamAppProfile": "BIGTABLE_CHANGE_STREAM_APP_PROFILE",
       "bigtableReadInstanceId": "BIGTABLE_READ_INSTANCE_ID",
       "bigtableReadTableId": "BIGTABLE_READ_TABLE_ID",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Bigtable_Change_Streams_to_Vector_Search",
     "environment": { "maxWorkers": "10" }
  }
}

Ganti kode berikut:

• PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
• JOB_NAME: nama pekerjaan unik pilihan Anda
• VERSION: versi {i>template<i} yang ingin Anda gunakan.

  Anda dapat menggunakan nilai berikut:

  • latest untuk menggunakan template versi terbaru, yang tersedia di Folder induk non-dated di bucket— gs://dataflow-templates-REGION_NAME/latest/
  • nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi tertentu dari , yang dapat ditemukan disarangkan di masing-masing folder induk bertanggal dalam bucket— gs://dataflow-templates-REGION_NAME/
  Perhatian: Versi template terbaru mungkin diupdate dengan menyebabkan gangguan perubahan. Lingkungan produksi Anda harus menggunakan template yang disimpan dalam versi terbaru dated untuk mencegah perubahan yang dapat menyebabkan gangguan ini memengaruhi di alur kerja produksi.
• LOCATION: wilayah tempat Anda ingin men-deploy tugas Dataflow Anda—misalnya, us-central1
• EMBEDDING_COLUMN: kolom Penyematan
• EMBEDDING_BYTE_SIZE: Ukuran byte dari array embedding. Bisa 4 atau 8.
• VECTOR_SEARCH_INDEX: Jalur indeks Penelusuran Vektor
• BIGTABLE_CHANGE_STREAM_APP_PROFILE: ID profil aplikasi Bigtable
• BIGTABLE_READ_INSTANCE_ID: ID Instance Bigtable sumber
• BIGTABLE_READ_TABLE_ID: ID tabel Bigtable sumber

Kode sumber template

Java

/*
 * Copyright (C) 2024 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates.bigtablechangestreamstovectorsearch;

import com.google.cloud.Timestamp;
import com.google.cloud.aiplatform.v1.IndexDatapoint;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadChangeStreamOptions;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions.ReadOptions;
import com.google.cloud.teleport.v2.cdc.dlq.DeadLetterQueueManager;
import com.google.cloud.teleport.v2.options.BigtableChangeStreamsToVectorSearchOptions;
import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
import com.google.cloud.teleport.v2.utils.DurationUtils;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.bigtable.BigtableIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.GroupIntoBatches;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Values;
import org.apache.beam.sdk.transforms.WithKeys;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Template(
    name = "Bigtable_Change_Streams_to_Vector_Search",
    category = TemplateCategory.STREAMING,
    displayName = "Bigtable Change Streams to Vector Search",
    description =
        "Streaming pipeline. Streams Bigtable data change records and writes them into Vertex AI Vector Search using Dataflow Runner V2.",
    optionsClass = BigtableChangeStreamsToVectorSearchOptions.class,
    optionsOrder = {
      BigtableChangeStreamsToVectorSearchOptions.class,
      ReadChangeStreamOptions.class,
      ReadOptions.class
    },
    skipOptions = {
      "bigtableReadAppProfile",
      "bigtableAdditionalRetryCodes",
      "bigtableRpcAttemptTimeoutMs",
      "bigtableRpcTimeoutMs"
    },
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigtable-change-streams-to-vector-search",
    flexContainerName = "bigtable-changestreams-to-vector-search",
    contactInformation = "https://cloud.google.com/support",
    streaming = true)
public final class BigtableChangeStreamsToVectorSearch {
  private static final Logger LOG =
      LoggerFactory.getLogger(BigtableChangeStreamsToVectorSearch.class);

  private static final String USE_RUNNER_V2_EXPERIMENT = "use_runner_v2";

  /**
   * Main entry point for executing the pipeline.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) throws Exception {
    LOG.info("Starting replication from Cloud Bigtable Change Streams to Vector Search");

    BigtableChangeStreamsToVectorSearchOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigtableChangeStreamsToVectorSearchOptions.class);

    run(options);
  }

  public static PipelineResult run(BigtableChangeStreamsToVectorSearchOptions options)
      throws IOException {
    options.setStreaming(true);
    options.setEnableStreamingEngine(true);

    List<String> experiments = options.getExperiments();
    if (experiments == null) {
      experiments = new ArrayList<>();
    }
    boolean hasUseRunnerV2 = false;
    for (String experiment : experiments) {
      if (experiment.equalsIgnoreCase(USE_RUNNER_V2_EXPERIMENT)) {
        hasUseRunnerV2 = true;
        break;
      }
    }
    if (!hasUseRunnerV2) {
      experiments.add(USE_RUNNER_V2_EXPERIMENT);
    }
    options.setExperiments(experiments);

    Instant startTimestamp =
        options.getBigtableChangeStreamStartTimestamp().isEmpty()
            ? Instant.now()
            : toInstant(Timestamp.parseTimestamp(options.getBigtableChangeStreamStartTimestamp()));

    String bigtableProjectId = getBigtableProjectId(options);

    LOG.info("  - startTimestamp {}", startTimestamp);
    LOG.info("  - bigtableReadInstanceId {}", options.getBigtableReadInstanceId());
    LOG.info("  - bigtableReadTableId {}", options.getBigtableReadTableId());
    LOG.info("  - bigtableChangeStreamAppProfile {}", options.getBigtableChangeStreamAppProfile());
    LOG.info("  - embeddingColumn {}", options.getEmbeddingColumn());
    LOG.info("  - crowdingTagColumn {}", options.getCrowdingTagColumn());
    LOG.info("  - project {}", options.getProject());
    LOG.info("  - indexName {}", options.getVectorSearchIndex());

    String indexName = options.getVectorSearchIndex();

    String vertexRegion = Utils.extractRegionFromIndexName(indexName);
    String vertexEndpoint = vertexRegion + "-aiplatform.googleapis.com:443";

    final Pipeline pipeline = Pipeline.create(options);

    DeadLetterQueueManager dlqManager = buildDlqManager(options);

    BigtableIO.ReadChangeStream readChangeStream =
        BigtableIO.readChangeStream()
            .withChangeStreamName(options.getBigtableChangeStreamName())
            .withExistingPipelineOptions(
                options.getBigtableChangeStreamResume()
                    ? BigtableIO.ExistingPipelineOptions.RESUME_OR_FAIL
                    : BigtableIO.ExistingPipelineOptions.FAIL_IF_EXISTS)
            .withProjectId(bigtableProjectId)
            .withAppProfileId(options.getBigtableChangeStreamAppProfile())
            .withInstanceId(options.getBigtableReadInstanceId())
            .withTableId(options.getBigtableReadTableId())
            .withMetadataTableInstanceId(options.getBigtableChangeStreamMetadataInstanceId())
            .withMetadataTableTableId(options.getBigtableMetadataTableTableId())
            .withStartTime(startTimestamp);

    PCollectionTuple results =
        pipeline
            .apply("Read from Cloud Bigtable Change Streams", readChangeStream)
            .apply("Create Values", Values.create())
            .apply(
                "Converting to Vector Search Datapoints",
                ParDo.of(
                        new ChangeStreamMutationToDatapointOperationFn(
                            options.getEmbeddingColumn(),
                            options.getEmbeddingByteSize(),
                            options.getCrowdingTagColumn(),
                            Utils.parseColumnMapping(options.getAllowRestrictsMappings()),
                            Utils.parseColumnMapping(options.getDenyRestrictsMappings()),
                            Utils.parseColumnMapping(options.getIntNumericRestrictsMappings()),
                            Utils.parseColumnMapping(options.getFloatNumericRestrictsMappings()),
                            Utils.parseColumnMapping(options.getDoubleNumericRestrictsMappings())))
                    .withOutputTags(
                        ChangeStreamMutationToDatapointOperationFn.UPSERT_DATAPOINT_TAG,
                        TupleTagList.of(
                            ChangeStreamMutationToDatapointOperationFn.REMOVE_DATAPOINT_TAG)));
    results
        .get(ChangeStreamMutationToDatapointOperationFn.UPSERT_DATAPOINT_TAG)
        .apply("Add placeholer keys", WithKeys.of("placeholder"))
        .apply(
            "Batch Contents",
            GroupIntoBatches.<String, IndexDatapoint>ofSize(
                    bufferSizeOption(options.getUpsertMaxBatchSize()))
                .withMaxBufferingDuration(
                    bufferDurationOption(options.getUpsertMaxBufferDuration())))
        .apply("Map to Values", Values.create())
        .apply(
            "Upsert Datapoints to VectorSearch",
            ParDo.of(new UpsertDatapointsFn(vertexEndpoint, indexName)))
        .apply(
            "Write errors to DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    results
        .get(ChangeStreamMutationToDatapointOperationFn.REMOVE_DATAPOINT_TAG)
        .apply("Add placeholder keys", WithKeys.of("placeholer"))
        .apply(
            "Batch Contents",
            GroupIntoBatches.<String, String>ofSize(
                    bufferSizeOption(options.getDeleteMaxBatchSize()))
                .withMaxBufferingDuration(
                    bufferDurationOption(options.getDeleteMaxBufferDuration())))
        .apply("Map to Values", Values.create())
        .apply(
            "Remove Datapoints From VectorSearch",
            ParDo.of(new RemoveDatapointsFn(vertexEndpoint, indexName)))
        .apply(
            "Write errors to DLQ",
            DLQWriteTransform.WriteDLQ.newBuilder()
                .withDlqDirectory(dlqManager.getSevereDlqDirectory() + "YYYY/MM/dd/HH/mm/")
                .withTmpDirectory(dlqManager.getSevereDlqDirectory() + "tmp/")
                .setIncludePaneInfo(true)
                .build());

    return pipeline.run();
  }

  private static String getBigtableProjectId(BigtableChangeStreamsToVectorSearchOptions options) {
    return StringUtils.isEmpty(options.getBigtableReadProjectId())
        ? options.getProject()
        : options.getBigtableReadProjectId();
  }

  private static Instant toInstant(Timestamp timestamp) {
    if (timestamp == null) {
      return null;
    } else {
      return Instant.ofEpochMilli(timestamp.getSeconds() * 1000 + timestamp.getNanos() / 1000000);
    }
  }

  private static int bufferSizeOption(int size) {
    if (size < 1) {
      size = 1;
    }

    return size;
  }

  private static Duration bufferDurationOption(String duration) {
    if (duration.isEmpty()) {
      return Duration.standardSeconds(1);
    }

    return DurationUtils.parseDuration(duration);
  }

  private static DeadLetterQueueManager buildDlqManager(
      BigtableChangeStreamsToVectorSearchOptions options) {
    String dlqDirectory = options.getDlqDirectory();
    if (dlqDirectory.isEmpty()) {
      LOG.info("Falling back to temp dir for DLQ");

      String tempLocation = options.as(DataflowPipelineOptions.class).getTempLocation();

      LOG.info("Have temp location {}", tempLocation);
      if (tempLocation == null || tempLocation.isEmpty()) {
        tempLocation = "/";
      } else if (!tempLocation.endsWith("/")) {
        tempLocation += "/";
      }

      dlqDirectory = tempLocation + "dlq";
    }

    LOG.info("Writing dead letter queue to: {}", dlqDirectory);

    return DeadLetterQueueManager.create(dlqDirectory, 1);
  }
}