feat: Add the pubsub lite source and settings #17
Conversation
google-cla
bot
added
the
cla: yes
src/main/java/com/google/cloud/pubsublite/flink/PubsubLiteSource.java
Outdated
Show resolved
Hide resolved
| SourceReaderContext readerContext) throws Exception { | ||
| PubsubLiteDeserializationSchema<OUT> schema = settings.deserializationSchema(); | ||
| schema.open( | ||
| new DeserializationSchema.InitializationContext() { |
There was a problem hiding this comment.
does this not exist as a utility?
There was a problem hiding this comment.
I don't think so. At least I didn't find other sources using a utility to do this
| new PubsubLiteRecordEmitter<>(), | ||
| settings.getCursorCommitter(), | ||
| settings.getSplitReaderSupplier(), | ||
| new Configuration(), |
There was a problem hiding this comment.
Hmm is this correct? Is this supposed to be retrieved from somewhere?
There was a problem hiding this comment.
This is at least intentional
The only place the the configuration object is used is by the source reader base class here:
I don't really want to expose these settings to users, so I'm just passing in an empty config so we'll use the default value
src/main/java/com/google/cloud/pubsublite/flink/PubsubLiteSourceSettings.java
Outdated
Show resolved
Hide resolved
src/main/java/com/google/cloud/pubsublite/flink/PubsubLiteSourceSettings.java
Outdated
Show resolved
Hide resolved
Yeah, that makes sense. I'll move everything in a separate pr |
| @AutoValue | ||
| public abstract class PubsubLiteSourceSettings<OutputT> implements Serializable { | ||
| private static final Logger LOG = LoggerFactory.getLogger(PubsubLiteSourceSettings.class); | ||
| private static final long serialVersionUID = 0; |
There was a problem hiding this comment.
pick a "random" value. I.E. pound the number keys
|
|
||
| public abstract Boundedness boundedness(); | ||
|
|
||
| public abstract MessageTimestampExtractor timestampSelector(); |
There was a problem hiding this comment.
please add an // Optional comment on these that have defaults and add comments on each setting's getter and setter.
|
|
||
| public abstract FlowControlSettings flowControlSettings(); | ||
|
|
||
| public abstract Boundedness boundedness(); |
There was a problem hiding this comment.
please make this optional and unbounded by default?
Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
Fixes #<issue_number_goes_here> ☕️