[ANNOUNCE] Share your Streaming Stories with us at Current 2023

[Israel Ekpo]

Do you have a great data streaming story to share?

We want to hear from you!

Speaking at Current 2023 is a great way to connect with hundreds of your
peers, become more involved in the data streaming community, and have a
public platform for you to share your story of the future of streaming and
real-time data.

Check out the details at the link below and share your CFP with us

https://sessionize.com/current-2023/

Important Dates:

Call for Papers opens: January 12, 2023

Speaker Office Hours:
Wednesday, February 1: 10:00 - 11:00 GMT and  11:00 - 12:00 PST
Wednesday, March 8: 10:00 - 11:00 GMT and 11:00 - 12:00 PST
Wednesday, March 15: 10:00 - 11:00 GMT and 11:00 - 12:00 PST

Call for Papers closes: March 20, 2023

Speaker Notifications sent: April 18, 2023

Questions? Please email speak...@currentevent.io

We are waiting to hear from you.

Using pyflink from flink distribution

[Andrew Otto]

Hello,

I'm having quite a bit of trouble running pyflink from the default flink
distribution tarballs.  I'd expect the python examples to work as long as
python is installed, and we've got the distribution.  Some python
dependencies are not included in the flink distribution tarballs:
cloudpickle, py4j and pyflink are in opt/python.  Others are not, e.g.
protobuf.

Now that I'm looking, I see that the pyflink installation instructions

are
to install via pip.

I'm doing this in Docker for use with the flink-kubernetes-operator.  In
the Using Flink Python on Docker

instructions,
there is a pip3 install apache-flink step.  I find this strange, since I'd
expect the 'FROM flink:1.15.2'  part to be sufficient.

By pip installing apache-flink, this docker image will have the flink
distro installed at /opt/flink and FLINK_HOME set to /opt/flink
.
BUT ALSO flink lib jars will be installed at e.g.
/usr/local/lib/python3.7/dist-packages/pyflink/lib!
So, by following those instructions, flink is effectively installed twice
into the docker image.

Am I correct or am I missing something?

Is using pyflink from the flink distribution tarball (without pip) not a
supported way to use pyflink?

Thanks!
-Andrew Otto
 Wikimedia Foundation

Failed Checkpoints when Flink HPA enabled on Kubernetes

[Varun Narayanan Chakravarthy via user]

Hello Flink Users,
We have enabled Kubernetes HPA for our Flink applications (Task Manager
only). Our Flink Applications are running in Reactive Mode. When the
Kubernetes controller triggers a scale-up/scale-down of our job's Task
Managers, we get an alert for failed checkpoints. Interestingly enough, the
error is transient and the successful checkpoints continue to progress as
well. We want to identify what is causing these failed checkpoints during
HPA scale-up/scale-down activities. Below is an example error log during
checkpoint failure:

2022-12-13 05:08:22.339 [jobmanager-io-thread-1] INFO
 org.apache.flink.runtime.checkpoint.CheckpointCoordinator  - Completed
checkpoint 40393 for job  (488170 bytes,
checkpointDuration=2582 ms, finalizationTime=322 ms).
2022-12-13 05:08:28.083 [Checkpoint Timer] INFO
 org.apache.flink.runtime.checkpoint.CheckpointFailureManager  - Failed to
trigger checkpoint for job  since
Checkpoint triggering task Source: Custom Source -> Sink: Unnamed (1/79) of
job  is not being executed at the moment.
Aborting checkpoint. Failure reason: Not all required tasks are currently
running..
2022-12-13 05:09:19.437 [Checkpoint Timer] INFO
 org.apache.flink.runtime.checkpoint.CheckpointCoordinator  - Triggering
checkpoint 40394 (type=CheckpointType{name='Checkpoint',
sharingFilesStrategy=FORWARD_BACKWARD}) @ 1670908159435 for job
.
2022-12-13 05:09:25.208 [jobmanager-io-thread-1] INFO
 org.apache.flink.fs.s3.common.writer.S3Committer  - Committing
flink-ingest-sps-nv-consumer/2022-11-15T01:10:30Z//chk-40394/_metadata
with MPU ID
_3vKXSVBMuBM7207EpGvCXOTRQskAiPPj88DSTTn55Uzuc_76dnubmTAPBovyWbKBKU8Wxqz6SuFBJ8cZnAOH_PkGEP36KJzMFYYPmT.xZvmLnM.YX1oJSHN3VP1TXpJECY8y80psYvRWvbt2e8CMeoa9JiOWiGYGRmqLGRdlQA-
2022-12-13 05:09:25.747 [jobmanager-io-thread-1] INFO
 org.apache.flink.runtime.checkpoint.CheckpointCoordinator  - Completed
checkpoint 40394 for job  (482850 bytes,
checkpointDuration=5982 ms, finalizationTime=330 ms).

Varun

Re: Blob server connection problem

[Matthias Pohl via user]

We had issues like that in the past (e.g. FLINK-24923 [1], FLINK-10683
[2]). The error you're observing is caused by an unexpected byte being read
from the socket. The BlobServer protocol expects either 0 (for put
messages) or 1 (for get messages) being retrieved as a header for new
message blocks [3].
Reading different values might mean that there is some other process
sending data to the port the BlobServer is listening on. May you check your
network traffic?

Matthias

[1] https://issues.apache.org/jira/browse/FLINK-24923
[2] https://issues.apache.org/jira/browse/FLINK-10683
[3]
https://github.com/apache/flink/blob/ab264e4ab5a3bc6961a5128b1c7e19752508a7ca/flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobServerConnection.java#L115

On Fri, Jan 20, 2023 at 11:26 PM Yang Liu  wrote:

> Hello,
>
> Is anyone familiar with the "blob server connection"? We have constantly
> been seeing the "Error while executing Blob connection" error, which
> sometimes causes a job stuck in the middle of a run if there are too many
> connection errors and eventually causes a failure, though most of the time
> the streaming run mode can recover from that failure in the subsequent
> iterations of runs, but that slows down the entire process. We tried
> adjusting the blob.fetch.num-concurrent and some other blob parameters, but
> it was not very helpful, so we want to know what might be the root cause of
> the issue. Are there any Flink metrics or tools to help us monitor the blob
> server connections?
>
> We use:
>
>- Flink Kubernetes Operator
>- Flink 1.15.3 and 1.16.0
>- Kafka, filesystem(S3)
>- Hudi 0.11.1
>
> Full error message:
>
> java.io.IOException: Unknown operation 71
>   at 
> org.apache.flink.runtime.blob.BlobServerConnection.run(BlobServerConnection.java:116)
>  [flink-dist-1.15.3.jar:1.15.3]
> 2023-01-19 16:44:37,448 ERROR 
> org.apache.flink.runtime.blob.BlobServerConnection   [] - Error while 
> executing BLOB connection.
>
>
> Best regards,
> Yang
>

Re: Problem with custom SerializationSchema in Flink 1.15

[Chesnay Schepler]

It's a known issue that various connectors/wrappers/etc did not respect 
the schema lifecycle.


This was fixed in 1.16.0 in 
https://issues.apache.org/jira/browse/FLINK-28807.


You will have to lazily initialize the mapper in the serialize() method 
for previous versions.


On 24/01/2023 11:52, Peter Schrott wrote:

Hi Flink-User!


I recently updated a Flink job from Flink version 1.13 to 1.15 
(managed by AWS). The Flink Job is written in Java.


I found out that the Kinesis Producer was deprecated in favour of 
Kinesis Streams Sink [1]. When upgrading to the new sink I stumbled 
upon a problem withe a custom Serialisation Schema. I am using a 
custom implementation of the Serialisation Schema to deserialize 
result POJOs to JSON using Jacksons Object Mapper. This Object Mapper 
is initialised and set up in the open() method of the Serialisation 
Schema. The problem is, that this open method is not call intially.


I have not found any but report or indications towards this issue. Is 
this known or am I just “holding it wrong” (aka missing something)?


I created a minimal reproducible on my GitHub repo: 
https://github.com/peterschrott/flink-sink-open



Best & Thanks,
Peter


[1] 
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/#kinesis-producer

Problem with custom SerializationSchema in Flink 1.15

[Peter Schrott]

Hi Flink-User!


I recently updated a Flink job from Flink version 1.13 to 1.15 (managed by 
AWS). The Flink Job is written in Java.

I found out that the Kinesis Producer was deprecated in favour of Kinesis 
Streams Sink [1]. When upgrading to the new sink I stumbled upon a problem 
withe a custom Serialisation Schema. I am using a custom implementation of the 
Serialisation Schema to deserialize result POJOs to JSON using Jacksons Object 
Mapper. This Object Mapper is initialised and set up in the open() method of 
the Serialisation Schema. The problem is, that this open method is not call 
intially.

I have not found any but report or indications towards this issue. Is this 
known or am I just “holding it wrong” (aka missing something)?

I created a minimal reproducible on my GitHub repo: 
https://github.com/peterschrott/flink-sink-open


Best & Thanks,
Peter


[1] 
https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/#kinesis-producer

Using S3 as stream source in Flink

[Sriram Ganesh]

Hi Everyone,

I am thinking of switching my input source from Kafka to S3. First, I
couldn't find any streaming source connector for S3. I have some basic
questions about

1. How S3 will work as a streaming source with proper checkpointing.?
2. How Flink will manage the last offset processed from a file?
3. Is exactly_once possible while using S3 as a streaming source?
4. What could be the pros and cons of using an S3 kind of storage as a
streaming source?

Any help would be appreciated. Thanks in advance.

Thanks,
-- 
*Sriram G*
*Tech*