Feat. JetStream Support for Source

[somratdutta]

We have added support for NATS JetStream for Source.
Also added is a test-case for the Source class.
We are open to feedback and actively working on to improve the current state of this project.

[scottf]

What is this?

[somratdutta]

It was mistakenly pushed, we have removed this now.
Thanks for noticing.

[scottf]

I'm concerned about introducing yet another consumer config when there area already so many options to chose from in the library. Is there something we could make serializable in the client that would help?

[somratdutta]

Yes, I agree, we shouldn't be using so many options.
Further plan is to send a PR in nats-client library to make it Serialisable.

[somratdutta]

Since, the PR to make certain classes Serializable was rejected, I don't think we have other options.

[scottf]

I see this now, we will have to make something.

[scottf]

Here's what I'm doing. I'm changing the ConsumerConfiguration builder to accept json. That's just step one. This allows us to make a simple class to use:

public static class SerializableConsumerConfiguration implements Serializable {
    private static final long serialVersionUID = 1L;

    private String json;
    private transient ConsumerConfiguration cc;

    public SerializableConsumerConfiguration() {
        cc(ConsumerConfiguration.builder().build());
    }

    public SerializableConsumerConfiguration(ConsumerConfiguration cc) {
        cc(cc);
    }

    public SerializableConsumerConfiguration cc(ConsumerConfiguration cc) {
        this.cc = cc;
        json = cc.toJson();
        return this;
    }

    public String getJson() {
        return json;
    }

    public ConsumerConfiguration getConsumerConfiguration() {
        return cc;
    }

    private void writeObject(java.io.ObjectOutputStream out) throws IOException {
        out.defaultWriteObject();
    }

    private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        cc = ConsumerConfiguration.builder().json(json).build();
    }
}

[scottf]

I'm adding SerializableConsumerConfiguration directly to JNATS in the helpers section. Please review: nats-io/nats.java#1156

[somratdutta]

that's great @scottf, will review

[scottf]

Can DeserializationSchema be replaced with existing PayloadDeserializer? Or is there a specific reason you made this choice?

[somratdutta]

Flink has inbuilt implementation for Deserializer and Serializers.
In our application code, we are heavily invested in using Flink's native Deserializer (& Serializers) as our current Pulsar-Flink connector implements it. In my humble opinion, this would be the same for other folks moving from another messaging framework. Thoughts?

[scottf]

I don't think so. Consider the PayloadDeserializer interface

OutputT getObject(String subject, byte[] input, Headers headers);

versus the primary methods in the DeserializationSchema

T deserialize(byte[] var1) throws IOException;

The Nats message is more than just the data byte array. But there is your code could implement both interfaces and delegate the getObject implementation to call deserialize.

[scottf]

Can you use more descriptive names instead of T See existing source for examples

[somratdutta]

Thanks, will make the changes accordingly.

[scottf]

Can you add blank lines at end of files. https://stackoverflow.com/questions/2287967/why-is-it-recommended-to-have-empty-line-in-the-end-of-a-source-file

[somratdutta]

Added.

[scottf]

All source files must have the license. Look at any existing source file. This is a requirement of contributing to this specific repo.

[scottf]

I wonder if we should make this a different package, i.e. io.synadia.flink.jssource, or I suppose repackage the existing into io.synadia.flink.source.core and then have io.synadia.flink.source.js

[scottf]

Please use Nats instead NATS for class names

[somratdutta]

Okay, refactored to Nats

[somratdutta]

I totally agree with the idea of repackaging.
For an easy review of this PR, I will make those changes in the next PR as it would involve reviewing a lot of file changes.

[shiv4289]

Alternatively we could put it in a different commit if @scottf is fine to review by picking out commits individually.

[scottf]

This is just an initial pass, I still need to run validate.

[scottf]

Cannot repackage, this has been released and I do not want to major version.

[scottf]

No on this.

• We need to find a different name for this.
  • batch size is not a consumer configuration detail, it's a consume behavior detail.
  • it could be confusing with actual consumer configuration.
• Is there anything from the simplification api that we can match to?

[somratdnutanix]

Noted,
here's a list of suggested name for the above class, recommendation needed:

• NatsConsumeOptions
• NatsBehaviorSettings
• NatsRuntimeParameters
• NatsOperationalConfig
• NatsProcessingOptions
• NatsBehaviorConfig
• NatsDynamicConfig
• NatsPerformanceSettings
• NatsConsumeBehavior
• NatsConsumeParameters

Going for NatsConsumeOptions for now.

[scottf]

The JNats client does not use with in it's builders so i would prefer to be consistent with that.

[vu-hoang-phan-form3]

If I understand correctly, a split will be a NATS subject. I'm curious about whether Flink will spawn multiple source reader in parallel for a single split?

I'm asking because with Kafka, a split is a partition, and apparently only one consumer will read that partition at a time. NATS is different that multiple consumers can read from the same subject. If I need multiple readers to read a very big subject, how Flink will handle that?

[scottf]

Yes the subject is a split. Should only be one reader per split (per subject)

[vu-hoang-phan-form3]

If one has only one huge subject, one reader might not be enough to handle the rate. Usually we can spawn multiple nats client to read from the same subject but with this pattern there can only be one I believe.

[scottf]

I think we could try to figure out how to use multiple instances of the same pull consumer (works the same as push delivery group). So when that option is configured we would have to know the number of consumer instances and build the splits accordingly.

[sourabhaggrawal]

I agree that we should build support for multiple consumers reading from one subject. I will work on it.
However, for this PR, would prefer to limit it to one subject one consumer.
We can create an issue in this project as requirement for record keeping.

[sourabhaggrawal]

Hello @scottf, just to clear if I understand it right, you are suggesting that we should have :

1. one consumer for every subject under a regex subject like (ORDER.*) ?
2. one consumer for every subject from list of subjects that user will provide ?
3. multiple copies of one consumer to increase the read throughput, if subject is big ?

If I have understood it correctly you are suggesting to have multiple copies of one consumer, right ?

[scottf]

The original implementation made one split per subject filter. I'm suggesting that we need some sort of configuration hint that tell us to make N splits for a single subject filter. I suppose that could be accomplished by allowing the subject filter multiple times in the input list.
Also, I would not characterize a subject like "order.*" as a regex, it's a wildcard. As presented, it is simply a single subject filter to a consumer.

[sourabhaggrawal]

Yes order.* is one subject and thats how I see it too. Now I have got more clarity on this multiple consumer part which I was confusing myself with some other usecase.
Let me see it once again how we can fit creating multiple instance of same consumer for one subject.

[vu-hoang-phan-form3]

What if we don't need to have split at all? NATS supports multiple readers out-of-the-box already. So a NATS source just need to create multiple readers with the same config (same stream, same subject) and they will work in parallel without further config.

[souravagrawal]

Hi @scottf @vu-hoang-phan-form3 It does make sense to ignore split if we do not want to use checkpointing, which is the current implementation, and we are not making use of the Split.
But to implement the checkpointing we need Split, it holds the data which is fetched from source and same data needs to be provided back to Flink during snapshot.
I have changed the code to implement checkpoint here

souravagrawal@edfe8e2

Please go through with it once, I plan to merge it in somrat's fork soon.

[vu-hoang-phan-form3]

If a message is acknowledged but checkpointing fails, we lose the message.
I think message ack should be inside notifyCheckpointComplete method instead. Either that or inside snapshotState.

[scottf]

Agreed, ack should be done after checkpoint.

[somratdutta]

Sure, will resolve this.

[sourabhaggrawal]

Hi @vu-hoang-phan-form3, message is acknowledged in pollNext as we read, if for any reason checkpointing is failed after acknowledging the job for the message is anyways done, there is a chance of getting same message again if for some reason connector gets rebooted during message delivery and before it could ack the message, consumer will start redelivering message from where it stopped.
If you can help me understand how do we lose the message here if checkpointing is failed ?

[shiv4289]

@vu-hoang-phan-form3 you mean the processed data for the message might get lost?

[vu-hoang-phan-form3]

@shiv4289 yes
@sourabhaggrawal After processMessage here, the message is acknowledge and goes down the pipeline. Then if the next checkpoint fails, that same message will be lost.

There is indeed a chance of redelivering the same message twice if it's acknowledged in notifyCheckpointComplete. That method is best-effort anyways. But it's not a problem, deduplication can be implemented elsewhere. Plus, NATS is at-least-once delivery, so duplication can always happen.

[somratdutta]

Hi @scottf,
I have made some changes as per the latest discussion.
You can see the commit here.
I also agree on your PR for supporting Serializable Consumer Configuration, I would refactor the Jetstream connector code once its merged.

Upcoming improvements:

• add support for both Flink and Nats deserializers.
• implement checkpointing in JetstreamReader as per discussion.
• support for multiple subjects.

[somratdutta]

Updates in this commit :

• Removed NatsConsumeOptions and updated configurations in NatsJetStreamSource and related classes:

  • Replaced NatsConsumeOptions with SerializableConsumerConfiguration for more efficient JSON serialization.
  • Updated NatsJetStreamSource, NatsJetStreamSourceBuilder, and related classes to use SerializableConsumerConfiguration.
  • Ensured proper handling of ConsumerConfiguration instances through SerializableConsumerConfiguration.

• Fixed Method Visibility in ExampleErrorListener to Match Interface Requirements:

  • Adjusted method visibility in ExampleErrorListener to comply with the interface requirements.
  • Ensured all interface methods are properly implemented with the correct access modifiers.

• Refactored NatsJetStreamSourceTest:

  • Modified test cases to use SerializableConsumerConfiguration instead of NatsConsumeOptions.
  • Ensured proper thread interruption handling and Flink environment cleanup.
  • Added explicit null checks and descriptive comments for better code readability.

Note :
This refactoring is done based on this PR which adds Serialisable wrappers. It hasn't been released yet, so the jnats version is bumped for next release.

@scottf