-
Notifications
You must be signed in to change notification settings - Fork 41
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
* feat: Bigquery sink using depot (#154) * chore: fix checkstyle * feat: Checkpointing on bq sink (#187) * feat: prepare for commit * fix: clear the messages for pushing to bq * docs: add documentation for BQ sink in Dagger - [#188] * chore: version bump of depot Co-authored-by: Sumit Aich <[email protected]>
- Loading branch information
1 parent
6133f7f
commit ba209e2
Showing
35 changed files
with
960 additions
and
214 deletions.
There are no files selected for viewing
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
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
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
43 changes: 43 additions & 0 deletions
43
...n/src/main/java/io/odpf/dagger/common/serde/proto/serialization/KafkaProtoSerializer.java
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package io.odpf.dagger.common.serde.proto.serialization; | ||
|
|
||
| import org.apache.flink.api.common.serialization.SerializationSchema.InitializationContext; | ||
| import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema; | ||
| import org.apache.flink.types.Row; | ||
|
|
||
| import io.odpf.dagger.common.exceptions.serde.DaggerSerializationException; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.Objects; | ||
|
|
||
| public class KafkaProtoSerializer implements KafkaRecordSerializationSchema<Row> { | ||
| private final String outputTopic; | ||
| private final ProtoSerializer protoSerializer; | ||
| private static final Logger LOGGER = LoggerFactory.getLogger("KafkaSink"); | ||
|
|
||
| public KafkaProtoSerializer(ProtoSerializer protoSerializer) { | ||
| this(protoSerializer, ""); | ||
| } | ||
|
|
||
| public KafkaProtoSerializer(ProtoSerializer protoSerializer, String outputTopic) { | ||
| this.protoSerializer = protoSerializer; | ||
| this.outputTopic = outputTopic; | ||
| } | ||
|
|
||
| @Override | ||
| public void open(InitializationContext context, KafkaSinkContext sinkContext) throws Exception { | ||
| KafkaRecordSerializationSchema.super.open(context, sinkContext); | ||
| } | ||
|
|
||
| @Override | ||
| public ProducerRecord<byte[], byte[]> serialize(Row row, KafkaSinkContext context, Long timestamp) { | ||
| if (Objects.isNull(outputTopic) || outputTopic.equals("")) { | ||
| throw new DaggerSerializationException("outputTopic is required"); | ||
| } | ||
| LOGGER.info("row to kafka: " + row); | ||
| byte[] key = protoSerializer.serializeKey(row); | ||
| byte[] message = protoSerializer.serializeValue(row); | ||
| return new ProducerRecord<>(outputTopic, key, message); | ||
| } | ||
| } |
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
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
53 changes: 53 additions & 0 deletions
53
...c/test/java/io/odpf/dagger/common/serde/proto/serialization/KafkaProtoSerializerTest.java
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| package io.odpf.dagger.common.serde.proto.serialization; | ||
|
|
||
| import io.odpf.dagger.common.exceptions.serde.DaggerSerializationException; | ||
| import org.apache.flink.types.Row; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
| import org.junit.Assert; | ||
| import org.junit.Test; | ||
| import org.mockito.Mockito; | ||
|
|
||
| import static org.junit.Assert.assertEquals; | ||
| import static org.junit.Assert.assertThrows; | ||
|
|
||
| public class KafkaProtoSerializerTest { | ||
|
|
||
| @Test | ||
| public void shouldSerializeIntoKafkaRecord() { | ||
| ProtoSerializer serializer = Mockito.mock(ProtoSerializer.class); | ||
| String outputTopic = "test"; | ||
| Row element = new Row(1); | ||
| element.setField(0, "testing"); | ||
| byte[] keyBytes = "key".getBytes(); | ||
| byte[] valueBytes = "value".getBytes(); | ||
| Mockito.when(serializer.serializeKey(element)).thenReturn(keyBytes); | ||
| Mockito.when(serializer.serializeValue(element)).thenReturn(valueBytes); | ||
| KafkaProtoSerializer kafkaProtoSerializer = new KafkaProtoSerializer(serializer, outputTopic); | ||
| ProducerRecord<byte[], byte[]> record = kafkaProtoSerializer.serialize(element, null, null); | ||
| ProducerRecord<byte[], byte[]> expectedRecord = new ProducerRecord<>("test", keyBytes, valueBytes); | ||
| Assert.assertEquals(expectedRecord, record); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldThrowExceptionWhenOutputTopicIsNullForSerializeMethod() { | ||
| ProtoSerializer serializer = Mockito.mock(ProtoSerializer.class); | ||
| KafkaProtoSerializer kafkaProtoSerializer = new KafkaProtoSerializer(serializer, null); | ||
| Row element = new Row(1); | ||
| element.setField(0, "1234"); | ||
| DaggerSerializationException exception = assertThrows(DaggerSerializationException.class, | ||
| () -> kafkaProtoSerializer.serialize(element, null, System.currentTimeMillis() / 1000)); | ||
| assertEquals("outputTopic is required", exception.getMessage()); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldThrowExceptionWhenOutputTopicIsEmptyForSerializeMethod() { | ||
| ProtoSerializer serializer = Mockito.mock(ProtoSerializer.class); | ||
| KafkaProtoSerializer kafkaProtoSerializer = new KafkaProtoSerializer(serializer, ""); | ||
| Row element = new Row(1); | ||
| element.setField(0, "1234"); | ||
|
|
||
| DaggerSerializationException exception = assertThrows(DaggerSerializationException.class, | ||
| () -> kafkaProtoSerializer.serialize(element, null, System.currentTimeMillis() / 1000)); | ||
| assertEquals("outputTopic is required", exception.getMessage()); | ||
| } | ||
| } |
Oops, something went wrong.