Add Kafka sink DLQ for kafka to BQ template, kafka to gcs template #1570
+627
−133
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
AnandInguva
commented
May 16, 2024
v2/common/pom.xml
Outdated
| <dependency> | ||
| <groupId>org.apache.kafka</groupId> | ||
| <artifactId>kafka-clients</artifactId> | ||
| <version>3.7.0</version> |
There was a problem hiding this comment.
todo: Check the latest version?
There was a problem hiding this comment.
Remove these file changes
AnandInguva
force-pushed
the
kafka-dlq-sink
branch
from
a5eff0c
to
035e6ef
Compare
|
cc: @johnjcasey |
AnandInguva
commented
May 16, 2024
...ka-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/KafkaToBigQueryFlex.java
Show resolved
Hide resolved
Dippatel98
reviewed
May 16, 2024
...-to-bigquery/src/main/java/com/google/cloud/teleport/v2/transforms/AvroDynamicTransform.java
Outdated
Show resolved
Hide resolved
Dippatel98
reviewed
May 16, 2024
...ka-to-bigquery/src/main/java/com/google/cloud/teleport/v2/templates/KafkaToBigQueryFlex.java
Outdated
Show resolved
Hide resolved
an2x
requested changes
May 17, 2024
v2/common/pom.xml
Outdated
| </dependency> | ||
| <dependency> | ||
| <groupId>org.apache.kafka</groupId> | ||
| <artifactId>kafka-clients</artifactId> |
There was a problem hiding this comment.
v2/common module shouldn't have a dependency on Kafka, it's supposed to provide only generic classes/utils. Kafka-related code should be placed under v2/kafka-common, which already has Kafka client deps.
v2/common/src/main/java/com/google/cloud/teleport/v2/dlq/DlqUtils.java
Outdated
Show resolved
Hide resolved
v2/common/src/main/java/com/google/cloud/teleport/v2/dlq/DeadLetterQueueOptions.java
Outdated
Show resolved
Hide resolved
| /** | ||
| * {@link DeadLetterQueueOptions} is used for any Dead letter queue sinks for the failed records. | ||
| */ | ||
| public interface DeadLetterQueueOptions extends PipelineOptions { |
There was a problem hiding this comment.
If this Options class is expected to have only Kafka DLQ options, I'd call it KafkaDeadLetterQueueOptions.
v2/common/src/main/java/com/google/cloud/teleport/v2/dlq/KafkaDeadLetterQueue.java
Outdated
Show resolved
Hide resolved
| @@ -58,6 +70,9 @@ public abstract class AvroWriteTransform | |||
| PCollection<KafkaRecord<byte[], byte[]>>, WriteFilesResult<AvroDestination>> { | |||
| private static final String subject = "UNUSED"; | |||
| static final int DEFAULT_CACHE_CAPACITY = 1000; | |||
Comment on lines
257
to
259
| private String schemaRegistryURL = null; | ||
| private String schemaPath = null; | ||
| private boolean useMock; |
There was a problem hiding this comment.
Seems like all 3 can be made final?
There was a problem hiding this comment.
Getting error when changing this to final
v2/kafka-to-gcs/src/main/java/com/google/cloud/teleport/v2/transforms/AvroWriteTransform.java
Show resolved
Hide resolved
v2/kafka-to-gcs/src/main/java/com/google/cloud/teleport/v2/transforms/AvroWriteTransform.java
Show resolved
Hide resolved
| @@ -62,7 +68,8 @@ public POutput expand(PCollection<KafkaRecord<byte[], byte[]>> kafkaRecord) { | |||
| .setSchemaRegistryURL(options().getSchemaRegistryURL()) | |||
| .setSchemaPath(options().getSchemaPath()) | |||
| .setWindowDuration(options().getWindowDuration()) | |||
| .build()); | |||
| .build() | |||
| .withBadRecordErrorHandlers(errorHandlers())); | |||
There was a problem hiding this comment.
Calling withX() after build() seems like a strange pattern to me. Should setErrorHandlers() be part of the builder?
There was a problem hiding this comment.
I have only changed this(having setErrorHandlers) for the classes that have AutoValue annotations.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #1570 +/- ##
============================================
- Coverage 43.54% 40.86% -2.68%
- Complexity 654 2829 +2175
============================================
Files 294 745 +451
Lines 15542 43127 +27585
Branches 1548 4595 +3047
============================================
+ Hits 6768 17626 +10858
- Misses 8255 24002 +15747
- Partials 519 1499 +980
|
an2x
approved these changes
May 21, 2024
Dippatel98
reviewed
May 21, 2024
...common/src/main/java/com/google/cloud/teleport/v2/kafka/dlq/KafkaDeadLetterQueueOptions.java
Outdated
Show resolved
Hide resolved
This reverts commit 3d82b67.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
No description provided.