You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "tkaymak (via GitHub)" <gi...@apache.org> on 2026/03/03 14:22:05 UTC
[PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
tkaymak opened a new pull request, #37750:
URL: https://github.com/apache/beam/pull/37750
Implement withStartOffset for Debezium to handle restarts. (Addresses #28248)
------------------------
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3996052394
@Abacn thank you for comment. I think your solution sounds like a much better vision, and we should aim for that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-4068192361
@Abacn pleasure!
(I can't merge as I don't have the rights)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #37750:
URL: https://github.com/apache/beam/pull/37750
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-4054178556
@Abacn Thank you! Implemented the review comment and updated `CHANGES.md`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3993315872
@tvalentyn maybe @Abacn can have a look (?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-4038965857
> Sorry for the delay (back from OOO).
Absolutely no need to apologize.
I have adjusted and rebased according to your comments - very appreciated!
@gemini-code-assist summary
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3992847099
Hi @an2x @johnjcasey @Dippatel98 - could you please help find a reviewer for this? Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #37750:
URL: https://github.com/apache/beam/pull/37750#discussion_r2914285367
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.io.debezium;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.Map;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link OffsetRetainer} that persists the Debezium connector offset as a JSON file using Beam's
+ * {@link FileSystems} abstraction.
+ *
+ * <p>The {@code path} argument can point to any filesystem supported by the active Beam runner,
+ * including local disk, Google Cloud Storage, Amazon S3, and others
+ *
+ * <p>On every {@code task.commit()}, the latest offset is serialised to JSON and written to the
+ * given path (overwriting the previous file). On pipeline startup the file is read back and the
+ * connector resumes from the stored position. If the file does not yet exist the connector starts
+ * from the beginning of the change stream.
+ *
+ * <p>Example — resume from GCS:
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(
+ * new FileSystemOffsetRetainer("gs://my-bucket/debezium/orders-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p>Example — local filesystem (useful for testing):
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(new FileSystemOffsetRetainer("/tmp/debezium-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p><b>Note:</b> writes are not atomic. If the pipeline is killed mid-write, the offset file may
+ * be corrupt. In that case, delete the file and the connector will restart from the beginning.
+ */
+public class FileSystemOffsetRetainer implements OffsetRetainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemOffsetRetainer.class);
+ private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
+
+ private final String path;
+
+ // ObjectMapper is thread-safe after configuration and does not need to be serialised.
+ private transient @Nullable ObjectMapper objectMapper;
+
+ public FileSystemOffsetRetainer(String path) {
+ this.path = path;
+ }
+
+ private ObjectMapper mapper() {
+ if (objectMapper == null) {
+ objectMapper = new ObjectMapper();
+ }
+ return objectMapper;
+ }
+
+ /**
+ * Reads the offset JSON file and returns its contents, or {@code null} if the file does not exist
+ * or cannot be read.
+ */
+ @Override
+ public @Nullable Map<String, Object> loadOffset() {
+ try {
+ ResourceId resourceId = FileSystems.matchNewResource(path, /* isDirectory= */ false);
+ try (ReadableByteChannel channel = FileSystems.open(resourceId);
+ InputStream stream = Channels.newInputStream(channel)) {
+ Map<String, Object> offset = mapper().readValue(stream, MAP_TYPE);
+ LOG.info("OffsetRetainer: loaded offset from {}: {}", path, offset);
+ return offset;
+ }
+ } catch (FileNotFoundException e) {
+ LOG.info("OffsetRetainer: no offset file found at {}; starting from the beginning.", path);
+ return null;
+ } catch (IOException e) {
+ LOG.warn(
+ "OffsetRetainer: failed to load offset from {}; starting from the beginning.", path, e);
+ return null;
+ }
+ }
+
+ /**
+ * Serialises {@code offset} to JSON and writes it to the configured path, overwriting any
+ * existing file. Errors are logged as warnings and swallowed so the pipeline continues.
+ */
+ @Override
+ public void saveOffset(Map<String, Object> offset) {
+ try {
+ ResourceId resourceId = FileSystems.matchNewResource(path, /* isDirectory= */ false);
Review Comment:
Should we write to a tmp file (e.g. `final/file.path.tmp`) then move to final path, in case pipeline gets crashed in the middle, then we get corrupted file and previous written offset is lost
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java:
##########
@@ -122,7 +122,21 @@ public class KafkaSourceConsumerFn<T> extends DoFn<Map<String, String>, T> {
@GetInitialRestriction
public OffsetHolder getInitialRestriction(@Element Map<String, String> unused)
throws IOException {
- return new OffsetHolder(null, null, null, spec.getMaxNumberOfRecords(), spec.getMaxTimeToRun());
+ Map<String, Object> initialOffset = null;
+
+ // Retainer takes precedence: it reflects the most recently committed position.
+ OffsetRetainer retainer = spec.getOffsetRetainer();
+ if (retainer != null) {
+ initialOffset = retainer.loadOffset();
+ }
+
+ // Fall back to the explicit one-time override when the retainer has no saved offset.
Review Comment:
+1 this precedence choice makes sense.
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.io.debezium;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.Map;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link OffsetRetainer} that persists the Debezium connector offset as a JSON file using Beam's
+ * {@link FileSystems} abstraction.
+ *
+ * <p>The {@code path} argument can point to any filesystem supported by the active Beam runner,
+ * including local disk, Google Cloud Storage, Amazon S3, and others
+ *
+ * <p>On every {@code task.commit()}, the latest offset is serialised to JSON and written to the
+ * given path (overwriting the previous file). On pipeline startup the file is read back and the
+ * connector resumes from the stored position. If the file does not yet exist the connector starts
+ * from the beginning of the change stream.
+ *
+ * <p>Example — resume from GCS:
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(
+ * new FileSystemOffsetRetainer("gs://my-bucket/debezium/orders-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p>Example — local filesystem (useful for testing):
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(new FileSystemOffsetRetainer("/tmp/debezium-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p><b>Note:</b> writes are not atomic. If the pipeline is killed mid-write, the offset file may
+ * be corrupt. In that case, delete the file and the connector will restart from the beginning.
+ */
+public class FileSystemOffsetRetainer implements OffsetRetainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemOffsetRetainer.class);
+ private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
+
+ private final String path;
+
+ // ObjectMapper is thread-safe after configuration and does not need to be serialised.
+ private transient @Nullable ObjectMapper objectMapper;
+
+ public FileSystemOffsetRetainer(String path) {
+ this.path = path;
+ }
+
+ private ObjectMapper mapper() {
+ if (objectMapper == null) {
+ objectMapper = new ObjectMapper();
+ }
+ return objectMapper;
+ }
+
+ /**
+ * Reads the offset JSON file and returns its contents, or {@code null} if the file does not exist
+ * or cannot be read.
+ */
+ @Override
+ public @Nullable Map<String, Object> loadOffset() {
+ try {
+ ResourceId resourceId = FileSystems.matchNewResource(path, /* isDirectory= */ false);
+ try (ReadableByteChannel channel = FileSystems.open(resourceId);
+ InputStream stream = Channels.newInputStream(channel)) {
+ Map<String, Object> offset = mapper().readValue(stream, MAP_TYPE);
+ LOG.info("OffsetRetainer: loaded offset from {}: {}", path, offset);
+ return offset;
+ }
+ } catch (FileNotFoundException e) {
+ LOG.info("OffsetRetainer: no offset file found at {}; starting from the beginning.", path);
+ return null;
+ } catch (IOException e) {
+ LOG.warn(
Review Comment:
It suggests the path cannot be read. I wonder if we should surface the Exception and fail in this case.
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.io.debezium;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.Map;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link OffsetRetainer} that persists the Debezium connector offset as a JSON file using Beam's
+ * {@link FileSystems} abstraction.
+ *
+ * <p>The {@code path} argument can point to any filesystem supported by the active Beam runner,
+ * including local disk, Google Cloud Storage, Amazon S3, and others
+ *
+ * <p>On every {@code task.commit()}, the latest offset is serialised to JSON and written to the
+ * given path (overwriting the previous file). On pipeline startup the file is read back and the
+ * connector resumes from the stored position. If the file does not yet exist the connector starts
+ * from the beginning of the change stream.
+ *
+ * <p>Example — resume from GCS:
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(
+ * new FileSystemOffsetRetainer("gs://my-bucket/debezium/orders-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p>Example — local filesystem (useful for testing):
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(new FileSystemOffsetRetainer("/tmp/debezium-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p><b>Note:</b> writes are not atomic. If the pipeline is killed mid-write, the offset file may
+ * be corrupt. In that case, delete the file and the connector will restart from the beginning.
+ */
+public class FileSystemOffsetRetainer implements OffsetRetainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemOffsetRetainer.class);
+ private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
+
+ private final String path;
+
+ // ObjectMapper is thread-safe after configuration and does not need to be serialised.
+ private transient @Nullable ObjectMapper objectMapper;
+
+ public FileSystemOffsetRetainer(String path) {
+ this.path = path;
+ }
+
+ private ObjectMapper mapper() {
+ if (objectMapper == null) {
+ objectMapper = new ObjectMapper();
+ }
+ return objectMapper;
+ }
+
+ /**
+ * Reads the offset JSON file and returns its contents, or {@code null} if the file does not exist
+ * or cannot be read.
+ */
+ @Override
+ public @Nullable Map<String, Object> loadOffset() {
+ try {
+ ResourceId resourceId = FileSystems.matchNewResource(path, /* isDirectory= */ false);
+ try (ReadableByteChannel channel = FileSystems.open(resourceId);
+ InputStream stream = Channels.newInputStream(channel)) {
+ Map<String, Object> offset = mapper().readValue(stream, MAP_TYPE);
+ LOG.info("OffsetRetainer: loaded offset from {}: {}", path, offset);
+ return offset;
+ }
+ } catch (FileNotFoundException e) {
+ LOG.info("OffsetRetainer: no offset file found at {}; starting from the beginning.", path);
+ return null;
+ } catch (IOException e) {
+ LOG.warn(
+ "OffsetRetainer: failed to load offset from {}; starting from the beginning.", path, e);
+ return null;
+ }
+ }
+
+ /**
+ * Serialises {@code offset} to JSON and writes it to the configured path, overwriting any
+ * existing file. Errors are logged as warnings and swallowed so the pipeline continues.
+ */
+ @Override
+ public void saveOffset(Map<String, Object> offset) {
+ try {
+ ResourceId resourceId = FileSystems.matchNewResource(path, /* isDirectory= */ false);
+ try (WritableByteChannel channel = FileSystems.create(resourceId, "application/json");
Review Comment:
Not sure how frequent a checkpoint gets issued, i.e. if we'll encounter similar issue like #37510. For now I wonder if we can compare the offset going to be saved with the last saved one, and only commit when it is modified, like what #37510 did
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3997933210
/gemini summary
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3992129923
Assigning reviewers:
R: @tvalentyn for label python.
R: @ahmedabu98 for label java.
R: @jrmccluskey for label go.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on code in PR #37750:
URL: https://github.com/apache/beam/pull/37750#discussion_r2917496868
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.io.debezium;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.Map;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link OffsetRetainer} that persists the Debezium connector offset as a JSON file using Beam's
+ * {@link FileSystems} abstraction.
+ *
+ * <p>The {@code path} argument can point to any filesystem supported by the active Beam runner,
+ * including local disk, Google Cloud Storage, Amazon S3, and others
+ *
+ * <p>On every {@code task.commit()}, the latest offset is serialised to JSON and written to the
+ * given path (overwriting the previous file). On pipeline startup the file is read back and the
+ * connector resumes from the stored position. If the file does not yet exist the connector starts
+ * from the beginning of the change stream.
+ *
+ * <p>Example — resume from GCS:
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(
+ * new FileSystemOffsetRetainer("gs://my-bucket/debezium/orders-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p>Example — local filesystem (useful for testing):
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(new FileSystemOffsetRetainer("/tmp/debezium-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p><b>Note:</b> writes are not atomic. If the pipeline is killed mid-write, the offset file may
+ * be corrupt. In that case, delete the file and the connector will restart from the beginning.
+ */
+public class FileSystemOffsetRetainer implements OffsetRetainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemOffsetRetainer.class);
+ private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
+
+ private final String path;
+
+ // ObjectMapper is thread-safe after configuration and does not need to be serialised.
+ private transient @Nullable ObjectMapper objectMapper;
+
+ public FileSystemOffsetRetainer(String path) {
Review Comment:
That is the Beam way of doing it, makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-4039161638
## Summary of Changes
This pull request introduces robust offset management capabilities to DebeziumIO, significantly enhancing its resilience and usability. It provides users with the flexibility to either explicitly define a starting offset for Debezium connectors or leverage an automatic file system-based mechanism to persist and restore offsets. This ensures seamless pipeline restarts and fault tolerance by allowing pipelines to resume data consumption from the last successfully processed point, preventing data loss and unnecessary reprocessing.
### Highlights
* **Comprehensive Offset Management**: Introduced both explicit `startOffset` and automatic `OffsetRetainer` mechanisms for DebeziumIO in Go, Java, and Python SDKs, allowing flexible control over data consumption resumption.
* **Automatic File-System Based Persistence**: Implemented a `FileSystemOffsetRetainer` in Java, enabling automatic saving and loading of Debezium connector offsets to any Beam-compatible filesystem (e.g., GCS, S3) after each successful checkpoint, ensuring fault tolerance with atomic writes.
* **Cross-Language API Consistency**: Ensured that the new offset management features are consistently exposed and configurable across the Go, Java, and Python SDKs, providing a unified experience for DebeziumIO users.
* **Robust Pipeline Restart Capabilities**: Significantly enhanced the resilience of Debezium-based pipelines by allowing them to gracefully restart from a known good state, preventing data loss or reprocessing of already handled events, with `OffsetRetainer` taking precedence over `startOffset`.
🧠 **New Feature in Public Preview:** You can now enable **Memory** to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
<details>
<summary><b>Changelog</b></summary>
* **sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go**
* Added `StartOffset` and `OffsetStoragePath` fields to the `readFromDebeziumSchema` struct.
* Implemented `StartOffset` and `OffsetStoragePath` functions as read options for configuring the Debezium connector.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java**
* Added abstract methods `getStartOffset`, `getOffsetRetainer`, `setStartOffset`, and `setOffsetRetainer` to the `Read` and `Builder` classes.
* Introduced `withStartOffset` and `withOffsetRetainer` builder methods with detailed Javadoc for configuring offset management.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java**
* Imported `HashMap` for map manipulation.
* Added `startOffset` and `offsetStoragePath` fields to the `Configuration` class and their corresponding setters.
* Implemented logic to parse `startOffset` strings into a `Map<String, Object>` and to configure `OffsetRetainer` based on `offsetStoragePath`.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java**
* Added a new class `FileSystemOffsetRetainer` to persist and load Debezium connector offsets as JSON files using Beam's `FileSystems`.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java**
* Modified `getInitialRestriction` to prioritize loading offsets from an `OffsetRetainer` or falling back to an explicit `startOffset`.
* Added logic to save the committed offset using the `OffsetRetainer` after each successful task commit.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/OffsetRetainer.java**
* Added a new `OffsetRetainer` interface for defining strategies to persist and restore Debezium connector offsets.
* **sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java**
* Added new test cases to verify the functionality of `withStartOffset` and `withOffsetRetainer`, including priority handling and null checks.
* **sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/FileSystemOffsetRetainerTest.java**
* Added unit tests for `FileSystemOffsetRetainer`, covering offset loading, saving, error handling, and serialization.
* **sdks/python/apache_beam/io/debezium.py**
* Extended `ReadFromDebeziumSchema` to include optional `start_offset` and `offset_storage_path` fields.
* Updated the `ReadFromDebezium` constructor to accept and document `start_offset` and `offset_storage_path` parameters.
</details>
<details>
<summary><b>Activity</b></summary>
* An initial summary of changes was provided by `gemini-code-assist[bot]`.
* Reviewers were assigned by `github-actions[bot]` to cover Python, Java, and Go changes.
* `tvalentyn` requested assistance in finding a reviewer for the PR.
* `tkaymak` suggested `@Abacn` as a potential reviewer.
* `Abacn` provided feedback, suggesting a more comprehensive `OffsetRetainer` interface for restartable offsets and atomic writes.
* `tkaymak` agreed with `Abacn`'s vision for a better solution.
* `tkaymak` requested a new summary from `gemini-code-assist[bot]`.
* `gemini-code-assist[bot]` provided an updated summary reflecting the new offset management features.
* `tkaymak` requested a second look from `@Abacn`.
* `tkaymak` adjusted and rebased the PR based on `Abacn`'s review comments.
* `Abacn` provided review comments on `FileSystemOffsetRetainer.java` regarding surfacing exceptions for unreadable files, using temporary files for atomic writes, skipping writes if the offset is unchanged, and using static factory methods. `tkaymak` acknowledged the suggestion for static methods.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3997936072
## Summary of Changes
This pull request significantly improves the resilience and usability of DebeziumIO by introducing robust offset management features. It allows users to either explicitly define a starting offset for Debezium connectors or leverage an automatic file system-based mechanism to persist and restore offsets, ensuring seamless pipeline restarts and fault tolerance.
### Highlights
* **Enhanced Offset Management**: Introduced `startOffset` and `OffsetStoragePath` capabilities for DebeziumIO in both Go and Java SDKs, allowing users to specify a point from which the Debezium connector should resume consuming changes or automatically manage offsets via a file system.
* **Automatic Offset Persistence**: Implemented an `OffsetRetainer` interface and a `FileSystemOffsetRetainer` class in Java, enabling automatic saving and loading of Debezium connector offsets to any Beam-compatible filesystem (e.g., GCS, S3) after each successful checkpoint.
* **Cross-Language Consistency**: Ensured that the new offset management features are consistently exposed and configurable across the Go, Java, and Python SDKs, providing a unified experience for DebeziumIO users.
* **Improved Pipeline Resilience**: Significantly enhanced the robustness of Debezium-based pipelines by allowing them to gracefully restart from a known good state, preventing data loss or reprocessing of already handled events.
🧠 **New Feature in Public Preview:** You can now enable **Memory** to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
<details>
<summary><b>Changelog</b></summary>
* **sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go**
* Added `StartOffset` and `OffsetStoragePath` fields to the `readFromDebeziumSchema` struct.
* Implemented new `StartOffset` and `OffsetStoragePath` functions as read options for configuring the Debezium connector.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java**
* Added abstract `getStartOffset`, `getOffsetRetainer`, `setStartOffset`, and `setOffsetRetainer` methods to the `Read` and `Builder` classes.
* Introduced `withStartOffset` and `withOffsetRetainer` builder methods with detailed Javadoc for configuring offset management.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java**
* Imported `HashMap` for map manipulation.
* Added `startOffset` and `offsetStoragePath` fields to the `Configuration` class and their corresponding setters.
* Implemented logic to parse `startOffset` strings into a `Map<String, Object>` and to configure `OffsetRetainer` based on `offsetStoragePath`.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java**
* Added a new class `FileSystemOffsetRetainer` to persist and load Debezium connector offsets as JSON files using Beam's `FileSystems`.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java**
* Modified `getInitialRestriction` to prioritize loading offsets from an `OffsetRetainer` or falling back to an explicit `startOffset`.
* Added logic to save the committed offset using the `OffsetRetainer` after each successful task commit.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/OffsetRetainer.java**
* Added a new `OffsetRetainer` interface for defining strategies to persist and restore Debezium connector offsets.
* **sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java**
* Added new test cases to verify the functionality of `withStartOffset` and `withOffsetRetainer`, including priority handling and null checks.
* **sdks/python/apache_beam/io/debezium.py**
* Extended `ReadFromDebeziumSchema` to include optional `start_offset` and `offset_storage_path` fields.
* Updated the `ReadFromDebezium` constructor to accept and document `start_offset` and `offset_storage_path` parameters.
</details>
<details>
<summary><b>Activity</b></summary>
* A summary of changes was provided by `gemini-code-assist[bot]`.
* Reviewers were assigned by `github-actions[bot]`.
* `tvalentyn` requested assistance in finding a reviewer.
* `tkaymak` suggested `Abacn` as a potential reviewer.
* `Abacn` provided feedback, suggesting a more comprehensive `OffsetRetainer` interface for restartable offsets.
* `tkaymak` agreed with `Abacn`'s vision for a better solution.
* `tkaymak` requested a new summary.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #37750:
URL: https://github.com/apache/beam/pull/37750#discussion_r2914352745
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/FileSystemOffsetRetainer.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.beam.io.debezium;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+import java.util.Map;
+import org.apache.beam.sdk.io.FileSystems;
+import org.apache.beam.sdk.io.fs.ResourceId;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An {@link OffsetRetainer} that persists the Debezium connector offset as a JSON file using Beam's
+ * {@link FileSystems} abstraction.
+ *
+ * <p>The {@code path} argument can point to any filesystem supported by the active Beam runner,
+ * including local disk, Google Cloud Storage, Amazon S3, and others
+ *
+ * <p>On every {@code task.commit()}, the latest offset is serialised to JSON and written to the
+ * given path (overwriting the previous file). On pipeline startup the file is read back and the
+ * connector resumes from the stored position. If the file does not yet exist the connector starts
+ * from the beginning of the change stream.
+ *
+ * <p>Example — resume from GCS:
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(
+ * new FileSystemOffsetRetainer("gs://my-bucket/debezium/orders-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p>Example — local filesystem (useful for testing):
+ *
+ * <pre>{@code
+ * DebeziumIO.read()
+ * .withConnectorConfiguration(config)
+ * .withOffsetRetainer(new FileSystemOffsetRetainer("/tmp/debezium-offset.json"))
+ * .withFormatFunction(myMapper);
+ * }</pre>
+ *
+ * <p><b>Note:</b> writes are not atomic. If the pipeline is killed mid-write, the offset file may
+ * be corrupt. In that case, delete the file and the connector will restart from the beginning.
+ */
+public class FileSystemOffsetRetainer implements OffsetRetainer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileSystemOffsetRetainer.class);
+ private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
+
+ private final String path;
+
+ // ObjectMapper is thread-safe after configuration and does not need to be serialised.
+ private transient @Nullable ObjectMapper objectMapper;
+
+ public FileSystemOffsetRetainer(String path) {
Review Comment:
nit: consider using static methods (e.g. `.of(path)`) to create objects
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on code in PR #37750:
URL: https://github.com/apache/beam/pull/37750#discussion_r2930135253
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java:
##########
@@ -123,6 +134,27 @@ public PTransform<PBegin, PCollection<String>> buildExternal(Configuration confi
readTransform = readTransform.withMaxTimeToRun(configuration.maxTimeToRun);
}
+ if (configuration.startOffset != null) {
+ Map<String, Object> startOffsetMap = new HashMap<>();
+ for (String property : configuration.startOffset) {
+ String[] parts = property.split("=", 2);
+ String key = parts[0];
+ String value = parts[1];
+ try {
+ startOffsetMap.put(key, Long.parseLong(value));
+ } catch (NumberFormatException e) {
+ startOffsetMap.put(key, value);
+ }
Review Comment:
Agree, implemented
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3995427363
Thanks, this change allows user to provide an offset that was obtained elsewhere. It now supports to read from an offset, yet not completely resolve #28248, as restarting the pipeline that have set the offset would still start from the same offset.
I had a (yet materialized) idea of restartable offset. Basically we can provide an "OffsetRetainer" interface to 1. read offset on pipeline startup; 2. write offset on checkpointing, and expose as a configuration for the IO. And we can then provide a built-in FileSystem based OffsetRetainer for user to use.
When pipeline is running, it continuously write committed offset to external location (gcs, kafka, etc) of choice; when pipeline starts or restarts, it tries to read from the same location at first.
If you prefer to go for current change for now, let me take a closer look for this PR. Or we can work on a more concrete solution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-4011888550
I would appreciate a second opinion on this @Abacn 🤝
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset for Kafka (#28248) [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-3991411095
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces a crucial `startOffset` capability to DebeziumIO, allowing users to define a specific point for Debezium connectors to begin or resume data consumption. This enhancement is vital for building resilient data pipelines that can recover from failures or planned shutdowns without losing data or re-processing unnecessary events. The implementation spans across the Go, Java, and Python SDKs, providing a unified experience for managing change data capture streams.
### Highlights
* **DebeziumIO Start Offset Implementation**: Introduced a `startOffset` feature for DebeziumIO in both Go and Java SDKs, allowing users to specify a point from which the Debezium connector should resume consuming changes.
* **Cross-language Compatibility**: The `startOffset` functionality is integrated across the Go, Java, and Python SDKs, ensuring consistent behavior for DebeziumIO users regardless of their chosen language.
* **Offset Parsing and Handling**: The Java SDK now includes logic to parse `startOffset` values provided as key-value strings, automatically converting numeric parts to `Long` where appropriate, facilitating flexible offset specification.
* **Improved Restart Capabilities**: This feature significantly enhances the robustness of Debezium-based pipelines by enabling them to gracefully restart from a known good state, preventing reprocessing of already handled data.
🧠 **New Feature in Public Preview:** You can now enable **Memory** to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
<details>
<summary><b>Changelog</b></summary>
* **sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go**
* Added `StartOffset` field to the `readFromDebeziumSchema` struct to hold offset configuration.
* Implemented a new `StartOffset` function that serves as a read option to configure the starting offset for the Debezium connector.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java**
* Added `getStartOffset` and `setStartOffset` abstract methods to the `Read` and `Builder` classes, respectively, to manage the new offset property.
* Introduced the `withStartOffset` builder method, allowing users to specify a map of key-value pairs representing the desired starting offset for the Debezium connector, complete with detailed Javadoc.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java**
* Imported `HashMap` for map manipulation.
* Added a `startOffset` field to the `Configuration` class and its corresponding setter method.
* Implemented logic within `buildExternal` to parse the list of `startOffset` strings into a `Map<String, Object>`, converting numeric values to `Long` where applicable, before applying it to the `readTransform`.
* **sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java**
* Modified the `getInitialRestriction` method to pass the newly introduced `startOffset` from the `spec` to the `OffsetHolder`.
* **sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java**
* Added static imports for `assertEquals` and `assertNull`.
* Imported `ImmutableMap` for easier map creation in tests.
* Included new test cases to verify that `withStartOffset` correctly stores the offset, handles null offsets, throws exceptions for invalid null input, and that `getInitialRestriction` properly utilizes the provided start offset.
* **sdks/python/apache_beam/io/debezium.py**
* Imported `Dict` and `Union` from `typing`.
* Extended `ReadFromDebeziumSchema` to include an optional `start_offset` field, defined as a list of strings.
* Updated the `ReadFromDebezium` constructor to accept a `start_offset` parameter and included comprehensive documentation for its usage and format.
* Passed the `start_offset` parameter to the `ReadFromDebeziumSchema` during initialization.
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
<b>You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the [Gemini Code Assist IDE Extension](https://cloud.google.com/products/gemini/code-assist).</b>
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #37750:
URL: https://github.com/apache/beam/pull/37750#issuecomment-4048285196
/gemini review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [DebeziumIO] Implement startOffset & offset persistence for Kafka (#28248) [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #37750:
URL: https://github.com/apache/beam/pull/37750#discussion_r2926012931
##########
sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java:
##########
@@ -123,6 +134,27 @@ public PTransform<PBegin, PCollection<String>> buildExternal(Configuration confi
readTransform = readTransform.withMaxTimeToRun(configuration.maxTimeToRun);
}
+ if (configuration.startOffset != null) {
+ Map<String, Object> startOffsetMap = new HashMap<>();
+ for (String property : configuration.startOffset) {
+ String[] parts = property.split("=", 2);
+ String key = parts[0];
+ String value = parts[1];
+ try {
+ startOffsetMap.put(key, Long.parseLong(value));
+ } catch (NumberFormatException e) {
+ startOffsetMap.put(key, value);
+ }
Review Comment:
The code assumes that each property string in `configuration.startOffset` will contain an `=` character. If a string without `=` is passed from a cross-language SDK, `property.split("=", 2)` will return an array of length 1, and accessing `parts[1]` will cause an `ArrayIndexOutOfBoundsException`. It would be more robust to validate the format of the property string to prevent the pipeline from crashing with an unexpected error.
```java
String[] parts = property.split("=", 2);
if (parts.length != 2) {
throw new IllegalArgumentException(
"Invalid startOffset property format. Expected 'key=value', but got: " + property);
}
String key = parts[0];
String value = parts[1];
try {
startOffsetMap.put(key, Long.parseLong(value));
} catch (NumberFormatException e) {
startOffsetMap.put(key, value);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org