How can I use my own custom sink connector based on AsyncSin Apache Flink #troubleshooting

How can I use my own custom sink connector based o...

chankyeong won

04/26/2023, 7:29 AM

How can I use my own custom sink connector based on AsyncSinkBase? I implemented components referenced this document, How to use

MySink

class on the DataStream in my flink application’s main method?

chankyeong won

04/26/2023, 7:33 AM

I found the answer at here. The answer is

sinkTo(new MySink(…))

method!

Ahmed Hamdy

04/26/2023, 4:14 PM

Glad you found the answer from our tests😄 feel free to reach out to me if you have any questions regarding AsyncSink 🙌

👍 1

chankyeong won

05/02/2023, 1:30 AM

@Ahmed Hamdy Hello. I have a question about AsyncSink element’s byte size. I need to implement

getSizeInBytes

function in AsyncSinkWriter class. I wonder how should I calculate element’s byte size. It should contain the java object size? My record(java pojo class) has the multiple data types for member variables, like String, Integer, Boolean, List<String, Boolean>, Map<String, String> and so on. What should I calculate for element’s byte size to exactly?

Ahmed Hamdy

05/02/2023, 8:10 AM

Hi @chankyeong won the

getSizeInBytes

is mainly used in 2 main use cases • flushing buffer of reaching

maxBufferSizeInBytes

• calculating the

numOfBytesOut

metric by sink. If you provide a state serializer for the sink, I would assume it would be easy to calculate number of bytes of each record after serializing it

chankyeong won

05/02/2023, 8:20 AM

I’m using the

SerializationUtils

Copy code

import my.model.Goods;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.connector.base.sink.writer.AsyncSinkWriterStateSerializer;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

@Internal
public class EsAsyncStateSerializer extends AsyncSinkWriterStateSerializer<Goods> {

    @Override
    protected void serializeRequestToStream(Goods request, DataOutputStream out) throws IOException {
        byte[] data = SerializationUtils.serialize(request);
        out.write(data);
    }

    @Override
    protected Goods deserializeRequestFromStream(long requestSize, DataInputStream in) throws IOException {
        byte[] data = new byte[(int) requestSize];
        in.read(data);

        return SerializationUtils.deserialize(data);
    }

    @Override
    public int getVersion() {
        return 1;
    }
}

Ahmed Hamdy

05/02/2023, 8:30 AM

yeah, you can use something similar to

SerializationUtils.serialize(request).length()

to get size in bytes then

chankyeong won

05/02/2023, 8:41 AM

I already tried. but error is occured.

org.apache.commons.lang3.SerializationException: java.io.NotSerializableException: org.apache.avro.generic.GenericData$Array

The source of this flink application is kafka with avro records. I think error reason is that all the data is processed within one task.

Ahmed Hamdy

05/02/2023, 12:00 PM

hmmm, this might be an issue with your Pojo class that it is not serializable. would you mind sharing full exception trace?

chankyeong won

05/04/2023, 9:54 AM

Copy code

org.apache.commons.lang3.SerializationException: java.io.NotSerializableException: org.apache.avro.generic.GenericData$Array
    at org.apache.commons.lang3.SerializationUtils.serialize(SerializationUtils.java:157)
    at org.apache.commons.lang3.SerializationUtils.serialize(SerializationUtils.java:179)
    at com.ably.sink.EsAsyncSinkWriter.getSizeInBytes(EsAsyncSinkWriter.java:104)
    at com.ably.sink.EsAsyncSinkWriter.getSizeInBytes(EsAsyncSinkWriter.java:32)
    at org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.addEntryToBuffer(AsyncSinkWriter.java:444)
    at org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.write(AsyncSinkWriter.java:327)
    at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.processElement(SinkWriterOperator.java:160)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
    at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
    at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
    at org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry.emitResult(StreamRecordQueueEntry.java:64)
    at org.apache.flink.streaming.api.operators.async.queue.UnorderedStreamElementQueue$Segment.emitCompleted(UnorderedStreamElementQueue.java:272)
    at org.apache.flink.streaming.api.operators.async.queue.UnorderedStreamElementQueue.emitCompletedElement(UnorderedStreamElementQueue.java:159)
    at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.outputCompletedElement(AsyncWaitOperator.java:389)
    at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.access$1200(AsyncWaitOperator.java:91)
    at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.processResults(AsyncWaitOperator.java:588)
    at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.lambda$processInMailbox$0(AsyncWaitOperator.java:569)
    at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
    at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
    at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.io.NotSerializableException: org.apache.avro.generic.GenericData$Array
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1185)
    at java.base/java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:349)
    at java.base/java.util.HashMap.internalWriteEntries(HashMap.java:1840)
    at java.base/java.util.HashMap.writeObject(HashMap.java:1411)
    at java.base/jdk.internal.reflect.GeneratedMethodAccessor2.invoke(Unknown Source)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at java.base/java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1145)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1497)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1553)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1510)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:349)
    at org.apache.commons.lang3.SerializationUtils.serialize(SerializationUtils.java:154)
    ... 32 more

chankyeong won

05/04/2023, 9:56 AM

My model is cassandra’s entity.

Copy code

@Table(keyspace = "ks", name = "goods")
public class Goods implements Serializable {

    @Transient
    private static final long serialVersionUID = 1L;
    ...
}

Ahmed Hamdy

05/04/2023, 10:48 AM

Yeah I think it just complains the model is not serializable, you would probably have the same issue when building the state. you can implement the

getSizeInBytes

separately if you want as I mentioned it is mainly used to flush the buffer on reaching

maxBufferSizeInBytes

and also reporting

numBytesOut

metric, but I would guess the serialziation issue would be there for state serialization anyway

chankyeong won

05/08/2023, 4:42 AM

@Ahmed Hamdy So, Is it entirely up to me to decide whether or not to include the java object size in the

getSizeInBytes

implementation?

Ahmed Hamdy

05/08/2023, 10:24 AM

@chankyeong won yeah as I mentioned, if you don't implement it (i.e. say return 0;) you will lose these 2 features in your implementation (byte size flushing and bytes out metricss)

Ahmed Hamdy

05/08/2023, 10:24 AM

otherwise your sink should work fine

chankyeong won

05/08/2023, 10:25 AM

I understand. Thank you!

9 Views

Open in Slack

Previous Next