How to start

[michiel2002v]

Hey!
I have a question about how to use the connector. Below you can find my current code where for now I have taken as much as possible from the example from this repo.

package org.dataflows;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import io.nats.client.Connection;
import io.nats.client.ConnectionListener;
//import org.support;
import io.nats.client.Dispatcher;
import io.nats.client.Nats;
import io.nats.client.Options;
import io.synadia.flink.Constants;
import io.synadia.flink.Utils;
import io.synadia.flink.examples.support.ExampleConnectionListener;
import io.synadia.flink.examples.support.ExampleErrorListener;
import io.synadia.flink.examples.support.Publisher;
import io.synadia.flink.sink.NatsSink;
import io.synadia.flink.sink.NatsSinkBuilder;
import io.synadia.flink.source.NatsSource;
import io.synadia.flink.source.NatsSourceBuilder;
import java.util.Properties;

public class Stateless {
    public static void main(String[] args) throws Exception {
        // load properties from a file for example stateless.properties
        Properties props = Utils.loadPropertiesFromFile("src/main/resources/stateless.properties");

        // make a connection to publish and listen with
        // props has io.nats.client.url in it
        Connection nc = connect(props);

        // create source
        NatsSource<String> source = new NatsSourceBuilder<String>()
                .sourceProperties(props)
                .connectionProperties(props)
                .build();
        System.out.println(source);

        // create sink
        NatsSink<String> sink = new NatsSinkBuilder<String>()
                .sinkProperties(props)
                .connectionProperties(props)
                .build();
        System.out.println(sink);

        // setup and start flink
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        DataStream<String> dataStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "NatsSource");

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.seconds(5)));



        // Sink the processed messages into a file
        String outputPath = "/test"; // Specify the path to the output file
        dataStream.writeAsText(outputPath, FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        env.executeAsync("Example");
    }

    private static Connection connect(Properties props) throws Exception {
        Options options = new Options.Builder()
                .properties(props)
                .connectionListener(new ExampleConnectionListener())
                .errorListener(new ExampleErrorListener())
                .build();
        return Nats.connect(options);
    }

}

However, I am getting this error:

/opt/flink# ./bin/flink run -c org.dataflows.Stateless myproject/Java-1.0-SNAPSH
OT.jar
WARNING: Unknown module: jdk.compiler specified to --add-exports
WARNING: Unknown module: jdk.compiler specified to --add-exports
WARNING: Unknown module: jdk.compiler specified to --add-exports
WARNING: Unknown module: jdk.compiler specified to --add-exports
WARNING: Unknown module: jdk.compiler specified to --add-exports

------------------------------------------------------------
 The program finished with the following exception:

java.lang.RuntimeException: Could not look up the main(String[]) method from the class org.dataflows.Stateless: io/nats/client/ConnectionListener
        at org.apache.flink.client.program.PackagedProgram.hasMainMethod(PackagedProgram.java:315)
        at org.apache.flink.client.program.PackagedProgram.<init>(PackagedProgram.java:161)
        at org.apache.flink.client.program.PackagedProgram.<init>(PackagedProgram.java:65)
        at org.apache.flink.client.program.PackagedProgram$Builder.build(PackagedProgram.java:691)
        at org.apache.flink.client.cli.CliFrontend.buildProgram(CliFrontend.java:1065)
        at org.apache.flink.client.cli.CliFrontend.getPackagedProgram(CliFrontend.java:272)
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
        at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1270)
        at org.apache.flink.client.cli.CliFrontend.lambda$mainInternal$10(CliFrontend.java:1367)
        at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
        at org.apache.flink.client.cli.CliFrontend.mainInternal(CliFrontend.java:1367)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1335)
Caused by: java.lang.NoClassDefFoundError: io/nats/client/ConnectionListener
        at java.base/java.lang.Class.getDeclaredMethods0(Native Method)
        at java.base/java.lang.Class.privateGetDeclaredMethods(Unknown Source)
        at java.base/java.lang.Class.getMethodsRecursive(Unknown Source)
        at java.base/java.lang.Class.getMethod0(Unknown Source)
        at java.base/java.lang.Class.getMethod(Unknown Source)
        at org.apache.flink.client.program.PackagedProgram.hasMainMethod(PackagedProgram.java:307)
        ... 11 more
Caused by: java.lang.ClassNotFoundException: io.nats.client.ConnectionListener
        at java.base/java.net.URLClassLoader.findClass(Unknown Source)
        at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67)
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51)
        at java.base/java.lang.ClassLoader.loadClass(Unknown Source)
        ... 17 more

My pom.xml looks like this.

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>vintecc.dataflows</groupId>
    <artifactId>NatsFlinkApp</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <!-- Add the flink-connector-nats dependency -->
        <dependency>
            <groupId>io.synadia</groupId>
            <artifactId>flink-connector-nats</artifactId>
            <version>1.0.0-beta1</version>
        </dependency>
        <!-- Jackson for JSON serialization/deserialization -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.17.0</version>
        </dependency>
        <!-- Apache Flink -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.17.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.17.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>2.0.9</version>
        </dependency>

        <!-- NATS -->
        <dependency>
            <groupId>io.nats</groupId>
            <artifactId>jnats</artifactId>
            <version>2.17.0</version>
        </dependency>
    </dependencies>

</project>

What could be going wrong?

Thanks in advance for helping me :)

[michiel2002v]

the problem is flink run not finding my dependencies
i solved it by building the jar as a fat jar

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>vintecc.dataflows</groupId>
    <artifactId>NatsFlinkApp</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <!-- Add the flink-connector-nats dependency -->
        <dependency>
            <groupId>io.synadia</groupId>
            <artifactId>flink-connector-nats</artifactId>
            <version>1.0.0-beta1</version>
        </dependency>
        <!-- Jackson for JSON serialization/deserialization -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.17.0</version>
        </dependency>
        <!-- Apache Flink -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.17.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.17.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-files</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.17.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>2.0.9</version>
        </dependency>

        <!-- NATS -->
        <dependency>
            <groupId>io.nats</groupId>
            <artifactId>jnats</artifactId>
            <version>2.17.0</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <!-- Add Maven Plugin to build fat jar from here https://jenkov.com/tutorials/maven/maven-build-fat-jar.html -->
            <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.1.1</version>

            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>

            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>

        </plugin>
        </plugins>
    </build>


</project>