Instantiating new ObjectMapper() for every single record in fromJson is extremely expensive and will cause severe performance degradation and high garbage collection overhead. Consider reusing a single, thread-safe static final ObjectMapper instance at the class level.

  public static Mod fromJson(String json) throws IOException {
    return OBJECT_MAPPER.readValue(json, Mod.class);
  }

I think this is worth implementing

Done

Instantiating new ObjectMapper() inside convertPropertiesToJson for every single cell mutation is highly inefficient. Please reuse a single static final ObjectMapper instance.

I think this is worth implementing

Done

options.getBigtableChangeStreamStartTimestamp() can return null if the parameter is not specified, which will cause a NullPointerException when calling .isEmpty(). Use StringUtils.isEmpty() to safely check for null or empty values.

+1

Done

Calling producer.send(record).get() synchronously blocks the pipeline thread for every single record, which severely limits throughput (e.g., to 1 / latency records per second per thread). To achieve high throughput while maintaining error handling, consider using asynchronous writes and buffering the futures, then blocking on them at the end of the bundle (or when a batch size is reached).

Maybe? If this is easy to implement, we can add it, otherwise we can hold off as is.

Done.
The integration test will follow.

I just pushed the integration test.

options.as(DataflowPipelineOptions.class).getTempLocation() can return null if not set, which will cause a NullPointerException when calling .endsWith("/"). Add a null check to handle this safely.

  private static DeadLetterQueueManager buildDlqManager(
      BigtableChangeStreamsToKafkaOptions options) {
    String tempLocation = options.as(DataflowPipelineOptions.class).getTempLocation();
    if (tempLocation != null) {
      tempLocation = tempLocation.endsWith("/") ? tempLocation : tempLocation + "/";
    } else {
      tempLocation = "";
    }
    String dlqDirectory =
        options.getDlqDirectory().isEmpty() ? tempLocation + "dlq/" : options.getDlqDirectory();

    LOG.info("DLQ directory: {}", dlqDirectory);
    return DeadLetterQueueManager.create(dlqDirectory, options.getDlqMaxRetries());
  }

+1

Done

Passing e.getCause() to SerializationException can discard the actual IOException context and stack trace if e.getCause() is null. Pass the exception e directly as the cause.

+1

Done

Splitting ignoreColumnFamilies and ignoreColumns using split("[\\s]*,[\\s]*") on strings with trailing commas or spaces can result in empty strings in the sets, which can cause bugs (e.g., ignoring empty column names). Filter out blank/empty strings after splitting.

    if (StringUtils.isBlank(ignoreColumnFamilies)) {
      this.columnFamiliesToIgnore = Collections.emptySet();
    } else {
      this.columnFamiliesToIgnore =
          Arrays.stream(ignoreColumnFamilies.split(","))
              .map(String::trim)
              .filter(StringUtils::isNotBlank)
              .collect(Collectors.toSet());
    }

    Set<String> columnsToIgnore;
    if (StringUtils.isBlank(ignoreColumns)) {
      columnsToIgnore = Collections.emptySet();
    } else {
      columnsToIgnore =
          Arrays.stream(ignoreColumns.split(","))
              .map(String::trim)
              .filter(StringUtils::isNotBlank)
              .collect(Collectors.toSet());
    }

+1

Done

This should probably work for Dataflow runner v1 (renamed to Streaming Java Runner now) as well. Can you please try ?

The integration tests passed with experiment disable_runner_v2 instead of use_runner_v2.

Update the PR/comment ?

Is this needed ? I would expect Beam to already register the String coder.

Done.

These are elements written via the same pipeline or other pipelines or both ?

Potentially, the user, by not providing dlqDirectory and tempLocation parameters can configure the DLQ to be shared across different runs or pipelines. We can add options.getJobName() to the DLQ path to avoid this risk. Would you like us to do it?

How about adding a random string to the path here ? Different jobs that are not run at the same time may share the job name.

Or did you intend the different runs of the same pipeline to share the DLQ ? In the case, job name makes sense.

Can this result in failed record being retried forever ? Usually we retry within the pipeline a given number of times potentially with backoff and then send the error records to DLQ for downstream processing (do not re-read from the same pipeline). Mostly I'm curious to know why you made this decision.

AFAIK this decision was part of the initial Google’s design. The number of attempted retries is stored, incremented and compared to the max limit. Please take a look at:

There's a special case when the user configures the number of retries to be 0 (which means: unlimited).
In our case the default value is 5, which should prevent from the failed records being retried forever.

I highly recommend using the official Beam Kafka sink here since there are lot more nuances when it comes to implementing a Beam sink which properly handles worker failures/retries, optimized for performance etc. Can the Beam Kafka sink be extended to add the missing functionality you need here ?

Another, more simpler, option might be to check for any potential errors that would not be captured by Beam KafkaIO DLQ [1] before the sink step in a DoFn and send them to DLQ manually.

[1] https://github.com/apache/beam/blob/69bdf5f29174adebe19006e1397acd6681ffce9d/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java#L1604

This was a conscious decision reached with Google, to use the hand-written simpler sink instead of KafkaIO. I think it was due to a mismatch between KafkaIO error handling and the requirements from the initial Google design.

@damccorm or @Abacn do you have additional context here ?

Could null values returned here and elsewhere result in serialization errors and send record to DLQ ?

Yes, there was a bug - fixed in 1e92e01 (note that it's still present in PubSubUtils on main branch!).

Apart from that, the nulls shouldn't be a problem - the Avro schema marks these fields as nullable.

Seems like we are double counting milliseconds here ?

Change to: getEpochSecond() * 1_000_000L + getNano() / 1_000

Done.

Note that this bug is still present in the Bigtable change streams to PubSub template on main.

Are these defined somewhere external that we can link to ?

It’s directly inspired by Avro schema, which is the definition of the external interface.

This remark led me to commit 5110126, which slightly cleaned the enum names up.

Done

Done

Done

Done