Cancelling a job with RMQSource
Hello, I have a flink streaming job that consumes data from RMQ source, process it and output it to RMQ sink. I want to delete the RMQ source queue when cancelling the job but keep it if the job failed to resume processing it's data when the job is restarted. One solution is to override the cancel() function to delete the queue but it will be deleted in both cases. Is the a way to just delete it in case of cancelling the job? Thanks in advance!
RE: Loading broadcast state on BroadcastProcessFunction instantiation or open method
Thank you, I have tried both approaches, Overriding open method did not work
but by implementing CheckpointedFunction and overriding initializeState I was
able to access and operate over broadcast state
@Override
public void initializeState(FunctionInitializationContext context) throws
Exception {
BroadcastState state =
context.getOperatorStateStore().getBroadcastState(RULES_DESCRIPTOR);
}
Regards,
From: Schwalbe Matthias
Sent: Tuesday, September 27, 2022 8:12 AM
To: David Anderson ; Vasquez, Alfredo
Cc: user@flink.apache.org
Subject: RE: Loading broadcast state on BroadcastProcessFunction instantiation
or open method
EXTERNAL MESSAGE
Hi Alfredo,
Did you consider implementing
org.apache.flink.streaming.api.checkpoint.CheckpointedFunction interface in
your broadcast function … the initializeState(…) function should give you
access to the state backend.
Kind regards
Thias
From: David Anderson mailto:dander...@apache.org>>
Sent: Tuesday, September 27, 2022 12:26 PM
To: alfredo.vasq...@spglobal.com
Cc: user@flink.apache.org
Subject: Re: Loading broadcast state on BroadcastProcessFunction instantiation
or open method
⚠EXTERNAL MESSAGE – CAUTION: Think Before You Click ⚠
Logically it would make sense to be able to initialize BroadcastState in the
open method of a BroadcastProcessFunction, but in practice I don't believe it
can be done -- because the necessary Context isn't made available.
Perhaps you could use the State Processor API to bootstrap some state into the
broadcast state.
David
On Mon, Sep 26, 2022 at 6:07 PM
alfredo.vasquez.spglobal.com via user
mailto:user@flink.apache.org>> wrote:
Hello community.
Currently we have a BroadcastProcessFunction implementation that is storing the
broadcast state using a MapStateDescriptor.
I have a use case that needs to load the BroadcastState to perform some
operation before receiving any processElement or processBroadcastElement
message.
Is there a way to load the BroadcastState on BroadcastProcessFunction
instantiation, overriding open(Configuration parameters) method or by
overriding some other callback function?
Kind regards,
The information contained in this message is intended only for the recipient,
and may be a confidential attorney-client communication or may otherwise be
privileged and confidential and protected from disclosure. If the reader of
this message is not the intended recipient, or an employee or agent responsible
for delivering this message to the intended recipient, please be aware that any
dissemination or copying of this communication is strictly prohibited. If you
have received this communication in error, please immediately notify us by
replying to the message and deleting it from your computer. S&P Global Inc.
reserves the right, subject to applicable local law, to monitor, review and
process the content of any electronic message or information sent to or from
S&P Global Inc. e-mail addresses without informing the sender or recipient of
the message. By sending electronic message or information to S&P Global Inc.
e-mail addresses you, as the sender, are consenting to S&P Global Inc.
processing any of your personal data therein.
Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und beinhaltet
unter Umständen vertrauliche Mitteilungen. Da die Vertraulichkeit von
e-Mail-Nachrichten nicht gewährleistet werden kann, übernehmen wir keine
Haftung für die Gewährung der Vertraulichkeit und Unversehrtheit dieser
Mitteilung. Bei irrtümlicher Zustellung bitten wir Sie um Benachrichtigung per
e-Mail und um Löschung dieser Nachricht sowie eventueller Anhänge. Jegliche
unberechtigte Verwendung oder Verbreitung dieser Informationen ist streng
verboten.
This message is intended only for the named recipient and may contain
confidential or privileged information. As the confidentiality of email
communication cannot be guaranteed, we do not accept any responsibility for the
confidentiality and the intactness of this message. If you have received it in
error, please advise the sender by return e-mail and delete this message and
any attachments. Any unauthorised use or dissemination of this information is
strictly prohibited.
Re: Is it practicle to enrich a Flink DataStream in middle operator with Flink Stateful Functions?
Hi Marco, The email is received by the list, but no answers have yet been provided unfortunately. Best regards, Martijn On Wed, Sep 28, 2022 at 4:11 PM Marco Villalobos wrote: > Did this list receive my email? > > I’m only asking because my last few questions have gone unanswered and > maybe the list server is blocking me. > > Anybody, please let me know. > > > On Sep 26, 2022, at 8:41 PM, Marco Villalobos > wrote: > > > > I indeed see the value of Flink Stateful Functions. > > > > However, if I already have a Flink Job, is it possible to enrich a > datastream with it? > > > > For example, like this: > > > > > > > > > > > > I really don't see how it would fit such a purpose. But, I do see that > it would be very at the end of a Flink Job, not enrichment, but more like a > handoff, more like this: > > > > > > > > Anybody care to eleborate on how to perform enrichment of a stream the > right way in Flink? > > I always find it problematic, and I was hoping Stateful Functions would > provide a silver bullet for that, but I think my thinking is incorrect. > > > > Thank you. > > > > Marco A. Villalobos >
Re: Is it practicle to enrich a Flink DataStream in middle operator with Flink Stateful Functions?
Did this list receive my email? I’m only asking because my last few questions have gone unanswered and maybe the list server is blocking me. Anybody, please let me know. > On Sep 26, 2022, at 8:41 PM, Marco Villalobos > wrote: > > I indeed see the value of Flink Stateful Functions. > > However, if I already have a Flink Job, is it possible to enrich a datastream > with it? > > For example, like this: > > > > > > I really don't see how it would fit such a purpose. But, I do see that it > would be very at the end of a Flink Job, not enrichment, but more like a > handoff, more like this: > > > > Anybody care to eleborate on how to perform enrichment of a stream the right > way in Flink? > I always find it problematic, and I was hoping Stateful Functions would > provide a silver bullet for that, but I think my thinking is incorrect. > > Thank you. > > Marco A. Villalobos
Re: [ANNOUNCE] Apache Flink 1.14.6 released
Thanks Xingbo for releasing it. Best, Jingsong On Wed, Sep 28, 2022 at 10:52 AM Xingbo Huang wrote: > > The Apache Flink community is very happy to announce the release of Apache > Flink 1.14.6, which is the fifth bugfix release for the Apache Flink 1.14 > series. > > Apache Flink® is an open-source stream processing framework for distributed, > high-performing, always-available, and accurate data streaming applications. > > The release is available for download at: > https://flink.apache.org/downloads.html > > Please check out the release blog post for an overview of the improvements > for this bugfix release: > https://flink.apache.org/news/2022/09/28/release-1.14.6.html > > The full release notes are available in Jira: > https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12315522&version=12351834 > > We would like to thank all contributors of the Apache Flink community who > made this release possible! > > Regards, > Xingbo
Re: Why is task manager shutting down?
Hi John Yes, the whole TaskManager exited because the task did not react to cancelling signal in time ``` 2022-08-30 09:14:22,138 ERROR org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Task did not exit gracefully within 180 + seconds. org.apache.flink.util.FlinkRuntimeException: Task did not exit gracefully within 180 + seconds. at org.apache.flink.runtime.taskmanager.Task$TaskCancelerWatchDog.run(Task.java:1791) [flink-dist_2.12-1.14.4.jar:1.14.4] at java.lang.Thread.run(Thread.java:750) [?:1.8.0_342] 2022-08-30 09:14:22,139 ERROR org.apache.flink.runtime.taskexecutor.TaskManagerRunner [] - Fatal error occurred while executing the TaskManager. Shutting it down... ``` And the task stack logged such as below when cancelling the sink task ``` 2022-08-30 09:14:22,135 WARN org.apache.flink.runtime.taskmanager.Task[] - Task 'Sink: jdbc (1/1)#359' did not react to cancelling signal - notifying TM; it is stuck for 180 seconds in method: java.net.SocketInputStream.socketRead0(Native Method) java.net.SocketInputStream.socketRead(SocketInputStream.java:116) java.net.SocketInputStream.read(SocketInputStream.java:171) java.net.SocketInputStream.read(SocketInputStream.java:141) com.microsoft.sqlserver.jdbc.TDSChannel.read(IOBuffer.java:2023) com.microsoft.sqlserver.jdbc.TDSReader.readPacket(IOBuffer.java:6418) com.microsoft.sqlserver.jdbc.TDSCommand.startResponse(IOBuffer.java:7579) com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.doExecutePreparedStatement(SQLServerPreparedStatement.java:592) com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement$PrepStmtExecCmd.doExecute(SQLServerPreparedStatement.java:524) com.microsoft.sqlserver.jdbc.TDSCommand.execute(IOBuffer.java:7194) com.microsoft.sqlserver.jdbc.SQLServerConnection.executeCommand(SQLServerConnection.java:2979) com.microsoft.sqlserver.jdbc.SQLServerStatement.executeCommand(SQLServerStatement.java:248) com.microsoft.sqlserver.jdbc.SQLServerStatement.executeStatement(SQLServerStatement.java:223) com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.execute(SQLServerPreparedStatement.java:505) com.xx.common.flink.connectors.jdbc.xxJdbcJsonOutputFormat.flush(xxJdbcJsonOutputFormat.java:111) com.xx.common.flink.connectors.jdbc.xxJdbcJsonSink.snapshotState(xxJdbcJsonSink.java:33) ``` Best, Congxian John Smith 于2022年9月23日周五 23:35写道： > Sorry new file: > https://www.dropbox.com/s/mm9521crwvevzgl/flink-flink-taskexecutor-274-flink-prod-v-task-0001.log?dl=0 > > On Fri, Sep 23, 2022 at 11:26 AM John Smith > wrote: > >> Hi I have attached the logs here... >> >> >> https://www.dropbox.com/s/12gwlps52lvxdhz/flink-flink-taskexecutor-274-flink-prod-v-task-0001.log?dl=0 >> >> 1- It looks like a timeout issue. Can someone confirm? >> 2- The task manager is restarted, since I have restart on failure in >> SystemD. But it seems after a few restarts it stops. Does it mean that >> SystemD has an internal counter of how many times it will restart a service >> before it doesn't do it anymore? >> >
Question regarding to debezium format
Hi guys, I was trying to use flink sql to consume data from kafka source, the format of which is debezium-avro-confluent. And I encountered a AvroTypeException saying that "Found something, expecting union", where something is not a type but a field that I defined in the schema registery. So I looked into the source code and what I found in org.apache.flink.formats.avro.registry.confluent.debezium.DebeziumAvroDeserializationSchema.createDebeziumAvroRowType was a segement of comment saying that "Debezium Avro contains other information, e.g. "source", "ts_ms", but we don't need them". I am wondering why don't we need it? Both source and ts_ms are in my schema, and I assume that the absence of source and ts_ms caused the exception described above. I am using Flink 1.15. Any help would be highly appreciated! Thanks!
