https://flink.apache.org/ logo
Join Slack
Powered by
Hi, we are implementing a flink job which is readi...
# troubleshooting
k
Hi, we are implementing a flink job which is reading events from RedPanda and applying rules using Table API, after applying rules we are using sql to sink to the destination topic
Copy code
private final String sink = "CREATE TABLE KafkaTable (\n" +
            "  `f0` STRING\n" +
            ") WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = 'result',\n" +
            "  'topic-pattern' = 'SINK_TABLE',\n" +
            "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
            "  'properties.group.id' = 'testGroup',\n" +
            "  'value.format' = 'raw'\n" +
            ");";
the issue here is that while inserting data to this table for sink it is taking very long time around 2-5s per event? so is there any better approach for the same
Copy code
resultTable.insertInto("KafkaTable").execute();
m
Which Flink version are you using?
k
1.17.1
m
With or without exactly once?
k
without exactly once
m
Is there anything in the logs? Are you sure it’s the sink, not in the join in the job?
k
Copy code
resultTable.insertInto("KafkaTable").execute();
this line is taking the most time, I have measured that and we are testing Table API and there is no join it is simply a pass through job
m
Yeah but that’s the part where basically all of the execution happens. You could check out the graph in the UI and then have a look at the metrics for the different operators
About the pass through: you must somehow link you rules to the source data, right? You can also check things like watermark progression, idleness etc. There isn’t yet really enough info to say where the problem lies
k
this is the flame graph for the job
the rules are broadcasted but I also have checked disabling the rules, just sinking the event without applying rule but its still taking that much time
m
What’s your input format source?
k
String
m
No, I mean AVRO, JSON, Protobuf, something else?
Also Raw?
k
Copy code
DataStream<String> dataStream = streamExecutionEnvironment.fromElements(data);

        streamTableEnvironment.createTemporaryView(
                "table",
                dataStream
        );
    private final String sink = "CREATE TABLE KafkaTable (\n" +
            "  `f0` STRING\n" +
            ") WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = 'result',\n" +
            "  'topic-pattern' = 'SINK_TABLE',\n" +
            "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
            "  'properties.group.id' = 'testGroup',\n" +
            "  'value.format' = 'raw'\n" +
            ");";

        // check for null
        Table resultTable = streamTableEnvironment.sqlQuery("SELECT * FROM table");

        streamTableEnvironment.executeSql(sql);

        resultTable.insertInto("KafkaTable").execute();

        streamTableEnvironment.dropTemporaryView(Main.TABLE_NAME);
        streamTableEnvironment.executeSql(drop);
here is the code for the processElement
yes Raw
m
Looking at the flamegraph, I’m wondering if a lot of time is spent on serializing and deserializing. Raw isn’t a really efficient format
k
then what should I do?
I also tried changing 
value.format
to 
json
from 
raw
but its still same result
m
That looks like a bunch of errors are thrown
Check the logs
k
yes you were write, flink is not able to find KafkaSink class, but dependency is bundled with jar not sure why is that, tried restarting flink but that doesnt seems to solve the issue
m
That implies that not all necessary dependencies are bundled in your JAR
k
The dependency is getting included in the jar I have confirmed that, can you verify my pom file if there is any problem in it
Copy code
<?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>org.example</groupId>
    <artifactId>test</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>
        <log4j.version>2.20.0</log4j.version>
    </properties>
    <dependencies>
        <!-- <https://mvnrepository.com/artifact/org.apache.flink/flink-java> -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.17.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka</artifactId>
            <version>1.17.1</version>
        </dependency>
        <!-- <https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kafka> -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-kafka</artifactId>
            <version>1.17.1</version>
        </dependency>


        <!-- <https://mvnrepository.com/artifact/org.apache.flink/flink-clients> -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.17.1</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.28</version>
            <scope>provided</scope>
        </dependency>

        <!-- <https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java> -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.17.1</version>
        </dependency>

        <!-- <https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-bridge> -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge</artifactId>
            <version>1.17.1</version>
        </dependency>

<!--        <dependency>-->
<!--            <groupId>org.apache.flink</groupId>-->
<!--            <artifactId>flink-table-planner_2.12</artifactId>-->
<!--            <version>1.17.1</version>-->
<!--        </dependency>-->

        <!--   Logging   -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>${log4j.version}</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.1.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <!-- Replace this with the main class of your job -->
                                    <mainClass>org.abs.security.Main</mainClass>
                                </transformer>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
I am still receiving this error in logs however Flink UI doesn't show any exception Caused by: org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user class: org.apache.flink.connector.kafka.sink.KafkaSink
m
You're including both flink-connector-kafka and flink-sql-connector-kafka: pick one of them, not both.
k
before I only had flink-connector-kafka at that time it was not working so i added flink-sql-connector-kafka, even with or without it the result was same
m
That's because they conflict with each other
If you're including flink-connector-kafka, then you also need to include the Kafka client dependencies yourself
Flink-sql-connector-kafka shades the Kafka client in
k
but is that the cause of this exception? because I can see KafkaSink getting bundled in jar
Copy code
Caused by: org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user class: org.apache.flink.connector.kafka.sink.KafkaSink
ClassLoader info: URL ClassLoader:
Class not resolvable through given classloader.
        at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:397)
        at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:166)
        at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.<init>(RegularOperatorChain.java:60)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:688)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:675)
        at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:952)
        at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:921)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:745)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.ClassNotFoundException: org.apache.flink.connector.kafka.sink.KafkaSink
        at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:589)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67)
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)
        at org.apache.flink.util.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:192)
        at java.base/java.lang.Class.forName0(Native Method)
        at java.base/java.lang.Class.forName(Class.java:398)
        at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76)
        at java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1995)
        at java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1862)
        at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2169)
        at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1679)
        at java.base/java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2464)
        at java.base/java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2358)
        at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2196)
        at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1679)
        at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:493)
        at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:451)
        at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:534)
        at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:522)
        at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:476)
        at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:383)
        ... 9 more
can someone help with this.
m
Please don't repeatedly send a message back to the Slack channel. There is something wrong with your bundling. I think you should be able to follow the instructions from the Flink documentation, or else inspect your JAR file to see what's actually in there.
3 Views
Open in Slack
PreviousNext
Powered by