-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #21 from synadia-io/jetstream-options
JetStreamOption support
- Loading branch information
Showing
47 changed files
with
481 additions
and
165 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
3 changes: 3 additions & 0 deletions
3
src/main/java/io/synadia/flink/v0/emitter/NatsRecordEmitter.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
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
2 changes: 1 addition & 1 deletion
2
src/main/java/io/synadia/flink/v0/enumerator/NatsSourceEnumeratorStateSerializer.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
2 changes: 1 addition & 1 deletion
2
src/main/java/io/synadia/flink/v0/enumerator/NatsSubjectSourceEnumeratorState.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
2 changes: 1 addition & 1 deletion
2
src/main/java/io/synadia/flink/v0/payload/PayloadDeserializer.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
2 changes: 1 addition & 1 deletion
2
src/main/java/io/synadia/flink/v0/payload/PayloadSerializer.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
2 changes: 1 addition & 1 deletion
2
src/main/java/io/synadia/flink/v0/payload/StringPayloadDeserializer.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
2 changes: 1 addition & 1 deletion
2
src/main/java/io/synadia/flink/v0/payload/StringPayloadSerializer.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
44 changes: 44 additions & 0 deletions
44
src/main/java/io/synadia/flink/v0/sink/NatsJetStreamSink.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,44 @@ | ||
| // Copyright (c) 2024 Synadia Communications Inc. All Rights Reserved. | ||
| // See LICENSE and NOTICE file for details. | ||
|
|
||
| package io.synadia.flink.v0.sink; | ||
|
|
||
| import io.synadia.flink.v0.payload.PayloadSerializer; | ||
| import io.synadia.flink.v0.sink.writer.NatsJetStreamSinkWriter; | ||
| import io.synadia.flink.v0.utils.ConnectionFactory; | ||
| import org.apache.flink.api.connector.sink2.SinkWriter; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * Flink Sink to publish data to one or more NATS subjects | ||
| * @param <InputT> the type of object from the source to convert for publishing | ||
| */ | ||
| public class NatsJetStreamSink<InputT> extends NatsSink<InputT> { | ||
|
|
||
| NatsJetStreamSink(List<String> subjects, | ||
| PayloadSerializer<InputT> payloadSerializer, | ||
| ConnectionFactory connectionFactory) | ||
| { | ||
| super(subjects, payloadSerializer, connectionFactory); | ||
| } | ||
|
|
||
| /** | ||
| * {@inheritDoc} | ||
| */ | ||
| @Override | ||
| public SinkWriter<InputT> createWriter(InitContext context) throws IOException { | ||
| return new NatsJetStreamSinkWriter<>(id, subjects, payloadSerializer, connectionFactory, context); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "NatsJetStreamSink{" + | ||
| "id='" + id + '\'' + | ||
| ", subjects=" + subjects + | ||
| ", payloadSerializer=" + payloadSerializer.getClass().getCanonicalName() + | ||
| ", connectionFactory=" + connectionFactory + | ||
| '}'; | ||
| } | ||
| } |
108 changes: 108 additions & 0 deletions
108
src/main/java/io/synadia/flink/v0/sink/NatsJetStreamSinkBuilder.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,108 @@ | ||
| // Copyright (c) 2024 Synadia Communications Inc. All Rights Reserved. | ||
| // See LICENSE and NOTICE file for details. | ||
|
|
||
| package io.synadia.flink.v0.sink; | ||
|
|
||
| import io.synadia.flink.v0.payload.PayloadSerializer; | ||
| import io.synadia.flink.v0.source.NatsSinkOrSourceBuilder; | ||
| import io.synadia.flink.v0.utils.Constants; | ||
| import io.synadia.flink.v0.utils.PropertiesUtils; | ||
|
|
||
| import java.util.Properties; | ||
|
|
||
| import static io.synadia.flink.v0.utils.Constants.PAYLOAD_SERIALIZER; | ||
| import static io.synadia.flink.v0.utils.Constants.SINK_PREFIX; | ||
|
|
||
| /** | ||
| * Builder to construct {@link NatsJetStreamSink}. | ||
| * | ||
| * <p>The following example shows the minimum setup to create a NatsSink that writes String values | ||
| * to one or more NATS subjects. | ||
| * | ||
| * <pre>{@code | ||
| * NatsJetStreamSink<String> sink = NatsJetStreamSink | ||
| * .<String>builder | ||
| * .subjects("subject1", "subject2") | ||
| * .connectionPropertiesFile("/path/to/jnats_client_connection.properties") | ||
| * .build(); | ||
| * }</pre> | ||
| * | ||
| * @see NatsSink | ||
| * @param <InputT> type of the records written | ||
| */ | ||
| public class NatsJetStreamSinkBuilder<InputT> extends NatsSinkOrSourceBuilder<NatsJetStreamSinkBuilder<InputT>> { | ||
| private PayloadSerializer<InputT> payloadSerializer; | ||
| private String payloadSerializerClass; | ||
|
|
||
| @Override | ||
| protected NatsJetStreamSinkBuilder<InputT> getThis() { | ||
| return this; | ||
| } | ||
|
|
||
| public NatsJetStreamSinkBuilder() { | ||
| super(SINK_PREFIX); | ||
| } | ||
|
|
||
| /** | ||
| * Set the payload serializer for the sink. | ||
| * @param payloadSerializer the serializer. | ||
| * @return the builder | ||
| */ | ||
| public NatsJetStreamSinkBuilder<InputT> payloadSerializer(PayloadSerializer<InputT> payloadSerializer) { | ||
| this.payloadSerializer = payloadSerializer; | ||
| this.payloadSerializerClass = null; | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Set the fully qualified name of the desired class payload serializer for the sink. | ||
| * @param payloadSerializerClass the serializer class name. | ||
| * @return the builder | ||
| */ | ||
| public NatsJetStreamSinkBuilder<InputT> payloadSerializerClass(String payloadSerializerClass) { | ||
| this.payloadSerializer = null; | ||
| this.payloadSerializerClass = payloadSerializerClass; | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Set sink properties from a properties object | ||
| * See the readme and {@link Constants} for property keys | ||
| * @param properties the properties object | ||
| * @return the builder | ||
| */ | ||
| public NatsJetStreamSinkBuilder<InputT> sinkProperties(Properties properties) { | ||
| baseProperties(properties); | ||
|
|
||
| String s = PropertiesUtils.getStringProperty(properties, PAYLOAD_SERIALIZER, prefixes); | ||
| if (s != null) { | ||
| payloadSerializerClass(s); | ||
| } | ||
|
|
||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Build a NatsSink. Subject and | ||
| * @return the sink | ||
| */ | ||
| public NatsJetStreamSink<InputT> build() { | ||
| if (payloadSerializer == null) { | ||
| if (payloadSerializerClass == null) { | ||
| throw new IllegalStateException("Valid payload serializer class must be provided."); | ||
| } | ||
|
|
||
| // so much can go wrong here... ClassNotFoundException, ClassCastException | ||
| try { | ||
| //noinspection unchecked | ||
| payloadSerializer = (PayloadSerializer<InputT>) Class.forName(payloadSerializerClass).getDeclaredConstructor().newInstance(); | ||
| } | ||
| catch (Exception e) { | ||
| throw new IllegalStateException("Valid payload serializer class must be provided.", e); | ||
| } | ||
| } | ||
|
|
||
| baseBuild(); | ||
| return new NatsJetStreamSink<>(subjects, payloadSerializer, createConnectionFactory()); | ||
| } | ||
| } |
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
50 changes: 50 additions & 0 deletions
50
src/main/java/io/synadia/flink/v0/sink/writer/NatsJetStreamSinkWriter.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,50 @@ | ||
| // Copyright (c) 2024 Synadia Communications Inc. All Rights Reserved. | ||
| // See LICENSE and NOTICE file for details. | ||
|
|
||
| package io.synadia.flink.v0.sink.writer; | ||
|
|
||
| import io.nats.client.JetStreamApiException; | ||
| import io.synadia.flink.v0.payload.PayloadSerializer; | ||
| import io.synadia.flink.v0.utils.ConnectionFactory; | ||
| import org.apache.flink.api.connector.sink2.Sink; | ||
| import org.apache.flink.util.FlinkRuntimeException; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.List; | ||
|
|
||
| /** | ||
| * This class is responsible to publish to one or more JetStream subjects | ||
| * @param <InputT> The type of the input elements. | ||
| */ | ||
| public class NatsJetStreamSinkWriter<InputT> extends NatsSinkWriter<InputT> { | ||
|
|
||
| public NatsJetStreamSinkWriter(String sinkId, | ||
| List<String> subjects, | ||
| PayloadSerializer<InputT> payloadSerializer, | ||
| ConnectionFactory connectionFactory, | ||
| Sink.InitContext sinkInitContext) throws IOException | ||
| { | ||
| super(sinkId, subjects, payloadSerializer, connectionFactory, sinkInitContext); | ||
| } | ||
|
|
||
| @Override | ||
| public void write(InputT element, Context context) throws IOException, InterruptedException { | ||
| byte[] payload = payloadSerializer.getBytes(element); | ||
| for (String subject : subjects) { | ||
| try { | ||
| ctx.js.publish(subject, null, payload); | ||
| } | ||
| catch (JetStreamApiException e) { | ||
| throw new FlinkRuntimeException(e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "NatsJetStreamSinkWriter{" + | ||
| "id='" + id + '\'' + | ||
| ", subjects=" + subjects + | ||
| '}'; | ||
| } | ||
| } |
Oops, something went wrong.