Hello,
Is new KafkaSource/KafkaSourceBuilder ready to be used ? If so, is KafkaSource
state compatible with legacy FlinkKafkaConsumer, for example if I replace
FlinkKafkaConsumer by KafkaSource, will offsets continue from what we had in
FlinkKafkaConsumer ?
Thanks,
Alexey
I'm receiving the following exception when trying to use a KafkaSource from
the new DataSource API.
Exception in thread "main" java.lang.NullPointerException
at
org.apache.flink.connector.kafka.source.reader.deserializer.ValueDeserializerWrapper.getProducedType(ValueDeserializer
Hello,
Our team tries to test reactive mode and replace FlinkKafkaConsumer with
the new KafkaSource.
But we can’t find the KafkaSource metrics list. Does anyone have any idea?
In our case, we want to know the Kafka consume delay and consume rate.
Thanks,
Oscar
On Tue, May 18, 2021 at 2:21 AM Alexey Trenikhun wrote:
> Hello,
>
> Is new KafkaSource/KafkaSourceBuilder ready to be used ? If so, is KafkaSource
> state compatible with legacy FlinkKafkaConsumer, for example if I replace
> FlinkKafkaConsumer
> by KafkaSource, will offsets
eceiving the following exception when trying to use a KafkaSource
> from the new DataSource API.
>
> Exception in thread "main" java.lang.NullPointerException
> at
> org.apache.flink.connector.kafka.source.reader.deserializer.ValueDeserializerWrapper.getProducedType(Value
m receiving the following exception when trying to use a KafkaSource
>> from the new DataSource API.
>>
>> Exception in thread "main" java.lang.NullPointerException
>> at
>> org.apache.flink.connector.kafka.source.read
>>
>> [1] https://issues.apache.org/jira/browse/FLINK-21691
>>
>> Cheers,
>> Till
>>
>> On Mon, Mar 8, 2021 at 3:35 PM Bobby Richard
>> wrote:
>>
>>> I'm receiving the following exception when trying to u
mode makes any difference.
On Mon, May 24, 2021 at 7:44 PM 陳樺威 wrote:
> Hello,
>
> Our team tries to test reactive mode and replace FlinkKafkaConsumer with
> the new KafkaSource.
> But we can’t find the KafkaSource metrics list. Does anyone have any idea?
> In our case, we want
Hi Ardhani,
Thanks for your kindly reply.
Our team use your provided metrics before, but the metrics disappear after
migrate to new KafkaSource.
We initialize KafkaSource in following code.
```
val consumer: KafkaSource[T] = KafkaSource.builder()
.setProperties(properties)
.setTopics(topic
Email: renqs...@gmail.com
On May 25, 2021, 2:35 PM +0800, 陳樺威 , wrote:
> Hi Ardhani,
>
> Thanks for your kindly reply.
>
> Our team use your provided metrics before, but the metrics disappear after
> migrate to new KafkaSource.
>
> We initialize KafkaSource in following
Looks like when KafkaSource is used instead of FlinkKafkaConsumer, metrics
listed below are not available. Bug? Work in progress?
Thanks,
Alexey
From: Ardhani Narasimha
Sent: Monday, May 24, 2021 9:08 AM
To: 陳樺威
Cc: user
Subject: Re: KafkaSource metrics
Use
Found https://issues.apache.org/jira/browse/FLINK-22766
From: Alexey Trenikhun
Sent: Tuesday, May 25, 2021 3:25 PM
To: Ardhani Narasimha ; 陳樺威
; Flink User Mail List
Subject: Re: KafkaSource metrics
Looks like when KafkaSource is used instead of
Hello all
I am confused.
What is the difference between KafkaSource as defined in :
https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/datastream/kafka/
and FlinkKafkaConsumer010 as defined in
https://nightlies.apache.org/flink/flink-docs-release-
1.2/api/java/org/apache
Hi,
Is there a way to run multiple flink jobs with the same Kafka group.id and have
them join the same consumer group?
It seems that setting the group.id using KafkaSource.builder().set_group_id()
does not have the effect of creating an actual consumer group in Kafka.
Running the same flink jo
Hi all,
I read these instructions for migrating to the KafkaSource:
https://nightlies.apache.org/flink/flink-docs-release-1.14/release-notes/flink-1.14/#deprecate-flinkkafkaconsumer
.
Do we need to employ any uid/allowNonRestoredState tricks if our Flink job
is also stateful outside of the
The latter link you posted refers to a very old flink release. You shold
use the first link, which refers to latest release
FG
On Tue, Feb 1, 2022 at 8:20 AM HG wrote:
> Hello all
>
> I am confused.
> What is the difference between KafkaSource as defined in :
> https://night
Hello Francesco
Perhaps I copied the wrong link of 1.2.
But there is also
https://nightlies.apache.org/flink/flink-docs-release-1.4/api/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.html
It seems there are 2 ways to use Kafka
KafkaSource source = KafkaSource.builder
I think the FlinkKakfaConsumer010 you're talking about is the old source
api. You should use only KafkaSource now, as they use the new source
infrastructure.
On Tue, Feb 1, 2022 at 9:02 AM HG wrote:
> Hello Francesco
> Perhaps I copied the wrong link of 1.2.
> But there i
in Flink 1.14 in favor of
KafkaSource, which implements the unified batch/streaming interface defined
in FLIP-27.
Regards,
David
On Tue, Feb 1, 2022 at 9:21 AM Francesco Guardiani
wrote:
> I think the FlinkKakfaConsumer010 you're talking about is the old source
> api. You shou
According to the Flink 1.12 documentation (
https://nightlies.apache.org/flink/flink-docs-release-1.12/dev/connectors/kafka.html),
it states to use FlinkKafkaSource when consuming from Kafka.
However, I noticed that the newer API uses KafkaSource, which uses
KafkaSourceBuilder and
Hi Robert,
This is a design choice. Flink's KafkaSource doesn't rely on consumer
groups for assigning partitions / rebalancing / offset tracking. It
manually assigns whatever partitions are in the specified topic across its
consumer instances, and rebalances only when the Flink job / Ka
Sorry, I meant to say "Hi Ben" :-)
On Thu, Mar 30, 2023 at 9:52 AM Tzu-Li (Gordon) Tai
wrote:
> Hi Robert,
>
> This is a design choice. Flink's KafkaSource doesn't rely on consumer
> groups for assigning partitions / rebalancing / offset tracking. It
> manua
Hi all,
I've been using the KafkaSource API as opposed to the classic consumer and
things have been going well. I configured my source such that it could be
used in either a streaming or bounded mode, with the bounded approach
specifically aimed at improving testing (unit/integration).
I would also be interested on instructions/discussion on how to state-migrate
from pre-unified sources/sinks to unified ones (Kafka) 😊
Thias
From: Mason Chen
Sent: Mittwoch, 27. Oktober 2021 01:52
To: user
Subject: FlinkKafkaConsumer -> KafkaSource State Migration
Hi all,
I read th
Hi,
Sorry for the late reply but most of use were involved in the Flink Forward
conference. The upgrade strategies for the Kafka sink and source are
pretty similar. Source and sink do not rely on state migration but leveraging
Kafka as source of truth.
When running with FlinkKafkaConsumer Maso
Thanks Fabian,
That was the information I was missing.
(Late reply ... same here, FlinkForward 😊 )
Thias
-Original Message-
From: Fabian Paul
Sent: Donnerstag, 28. Oktober 2021 08:38
To: Schwalbe Matthias
Cc: Mason Chen ; user
Subject: Re: FlinkKafkaConsumer -> KafkaSou
ation
(https://nightlies.apache.org/flink/flink-docs-release-1.12/dev/connectors/kafka.html),
it states to use FlinkKafkaSource when consuming from Kafka.
However, I noticed that the newer API uses KafkaSource, which uses
KafkaSourceBuilder and OffsetsInitializer.
Although I am on the Flink
guess we’ll need to just run it in a single
cluster and be aware of the risks if we lost that cluster.
Thanks,
Ben
On 2023/03/30 16:52:31 "Tzu-Li (Gordon) Tai" wrote:
> Hi Robert,
>
> This is a design choice. Flink's KafkaSource doesn't rely on consumer
>
Otherwise I guess we’ll need to just run it in
> a single cluster and be aware of the risks if we lost that cluster.
>
> Thanks,
> Ben
>
> On 2023/03/30 16:52:31 "Tzu-Li (Gordon) Tai" wrote:
> > Hi Robert,
> >
> > This is a design choice. Flin
would be streaming and
unbounded, however I’d love to have a reliable integration test or a pattern
that I could use to guarantee the processing of a finite set of data via a
KafkaSource (I.e. send finite records to Kafka, read from topic, process all
records, apply assertion after processing
at I could use to guarantee the processing of a finite set of
> data via a KafkaSource (I.e. send finite records to Kafka, read from topic,
> process all records, apply assertion after processing).
>
> Any ideas/recommendations/workarounds would be greatly welcome and I’d be
> happy to share
Hello,
When trying to reproduce a bug, we made a DeserialisationSchema that
throws an exception when a malformed message comes in.
Then, we sent a malformed message together with a number of well formed
messages to see what happens.
valsource= KafkaSource.builder[OurMessage]()
.setValueOnlyDe
Hi,
I am using a org.apache.flink.connector.kafka.source.KafkaSource with a
watermark strategy like this:
WatermarkStrategy.forMonotonousTimestamps().withIdleness(Duration.ofSeconds(10))
I noticed that after a short while all the partitions seem to be marked as
idle even though there are message
We've changed the KafkaSource to ingest from a new topic but the old name
is still being referenced:
2022-10-04 07:03:41org.apache.flink.util.FlinkException: Global
failure triggered by OperatorCoordinator for 'Source: Grokfailures'
(operator feca28aff5a3958840bee985e
Hi Team,
Greetings of the day..
we are on flink 1.16.1 version and using flinkkafkaconsumer today.
When I replaced it with kafkasource,it's failing with not able to connect
with kafka jaas configuration. Error says Kafka client entry not found in
/tmp/jass config file. We are passing the
I am using Flink 1.13.1 on AWS EMR and I seem to have hit this bug:
https://issues.apache.org/jira/browse/FLINK-27041. My job will fail when
there are empty partitions. I see it is fixed in a newer version of Flink
but I cannot update Flink version at this time.
Suggestions on a workaround? I a
Hey
We have a flinkapp which is subscribing to multiple topics, we recently
upgraded our application from 1.13 to 1.15, which we started to use
KafkaSource instead of FlinkKafkaConsumer (deprecated).
But we noticed some weird issue with KafkaSource after the upgrade, we are
setting the topics
Hi Frank,
This sounds like an interesting issue. Can you share a minimal working example?
Best regards,
Niklas
> On 9. Feb 2022, at 23:11, Frank Dekervel wrote:
>
> Hello,
>
> When trying to reproduce a bug, we made a DeserialisationSchema that throws
> an exception when a malformed message
Hi all,
After examining the source code further, I am quite sure
org.apache.flink.api.common.eventtime.WatermarksWithIdleness
does not work with FLIP-27 sources.
In org.apache.flink.streaming.api.operators.SourceOperator, there are
separate instances of WatermarksWithIdleness created for each spl
Although I'm not very familiar with the design of the code involved, I also
looked at the code, and I'm inclined to agree with you that this is a bug.
Please do raise an issue.
I'm wondering how you noticed this. I was thinking about how to write a
failing test, and I'm wondering if this has some
Thanks David,
I am working on a flink datastream job that does a temporal join of two
kafka topics based on watermarks. The problem was quite obvious when I
enabled idleness and data flowed through much faster with different results
even though the topics were not idle.
Regards.
On Mon, Aug 15,
Yan, I've created https://issues.apache.org/jira/browse/FLINK-28975 to
track this.
Regards,
David
On Sun, Aug 14, 2022 at 6:38 PM Yan Shen wrote:
> Thanks David,
>
> I am working on a flink datastream job that does a temporal join of two
> kafka topics based on watermarks. The problem was quite
Hello,
Is there way to configure Flink to expose watermarLag metric per topic per
partition? I think it could be useful to detect data skew between partitions
Thanks,
Alexey
Hi Robert,
Based on
https://stackoverflow.com/questions/72870074/apache-flink-restoring-state-from-checkpoint-with-changes-kafka-topic
I think you'll need to change the UID for your KafkaSource and restart your
job with allowNonRestoredState enabled.
Best regards,
Martijn
On Tue, Oct 4,
Hi Martjin,
I notice that this question comes up quite often. Would this be a good
addition to the KafkaSource documentation? I'd be happy to contribute to
the documentation.
Best,
Mason
On Tue, Oct 4, 2022 at 11:23 AM Martijn Visser
wrote:
> Hi Robert,
>
> Ba
Hi Mason,
Definitely! Feel free to open a PR and ping me for a review.
Cheers, Martijn
On Tue, Oct 4, 2022 at 3:51 PM Mason Chen wrote:
> Hi Martjin,
>
> I notice that this question comes up quite often. Would this be a good
> addition to the KafkaSource documentation? I&#
l free to open a PR and ping me for a review.
>
> Cheers, Martijn
>
> On Tue, Oct 4, 2022 at 3:51 PM Mason Chen wrote:
>
>> Hi Martjin,
>>
>> I notice that this question comes up quite often. Would this be a good
>> addition to the KafkaSource documentatio
rs, Martijn
>>
>> On Tue, Oct 4, 2022 at 3:51 PM Mason Chen wrote:
>>
>>> Hi Martjin,
>>>
>>> I notice that this question comes up quite often. Would this be a good
>>> addition to the KafkaSource documentation? I'd be happy to contribu
Hi,
I'm using a kafka source to read in messages from kafka into a datastream.
However I can't seem to access the key of the kafka message in the datastream.
Is this even possible ?
cheers
Noel
Hi Flink Community,
I have a few questions regarding the new KafkaSource and event time, which I
wasn't able to answer myself via checking the docs, but please point me to the
right pages in case I missed something. I'm not entirely whether my knowledge
entirely holds for the new K
> When I replaced it with kafkasource,it's failing with not able to connect
> with kafka jaas configuration. Error says Kafka client entry not found in
> /tmp/jass config file. We are passing the flonk runtime arg for the
> security.auth.login conf details. Same was working g w
it may be useful for
> positioning the issue
>
> Best,
> Shammon FY
>
>
> On Fri, Apr 21, 2023 at 12:56 AM naga sudhakar
> wrote:
>
>> Hi Team,
>> Greetings of the day..
>> we are on flink 1.16.1 version and using flinkkafkaconsumer today.
>> When
viewpage.action?pageId=217389320
On Wed, Nov 1, 2023 at 7:29 AM Emily Li via user wrote:
>
> Hey
>
> We have a flinkapp which is subscribing to multiple topics, we recently
> upgraded our application from 1.13 to 1.15, which we started to use
> KafkaSource instead of FlinkKafka
ng
this feature?
For our current situation, we are subscribing to hundreds of topics, and we
add/remove topics quite often (every few days probably), adding topics
seems to be okay at the moment, but with the current KafkaSource design, if
removing a topic means we need to change the kafka soure
's any plan in releasing
> this feature?
>
> For our current situation, we are subscribing to hundreds of topics, and
> we add/remove topics quite often (every few days probably), adding topics
> seems to be okay at the moment, but with the current KafkaSource design, if
> re
Using KafkaSource builder with a job parallelism larger than the number of
kafka partitions, the job is unable to checkpoint.
With a job parallelism of 4, 3 of the tasks are marked as FINISHED for the
kafka topic with one partition. For this reason checkpointing seems to be
disabled.
When using
The KafkaSource, and KafkaSourceBuilder appear to prevent users from
providing their own KafkaSubscriber. Am I overlooking something?
In my case I have an external system that controls which topics we should
be ingesting, and it can change over time. I need to add, and remove topics
as we refresh
data from a locally saved Parquet File, and a KafkaSource
consuming events from a remote Kafka broker.
I was wondering if instead of using a local Parquet file, whether it is
possible to directly stream the file from a GCS bucket and construct a File
Source out of it at runtime ? The Parquet Files
Hi Noel,
It's definitely possible. You need to implement a
custom KafkaRecordDeserializationSchema: its "deserialize" method gives you
a ConsumerRecord as an argument so that you can extract Kafka message key,
headers, timestamp, etc.
Then pass that when you create a
argument so that you can extract Kafka message key,
> headers, timestamp, etc.
>
> Then pass that when you create a KafkaSource via "setDeserializer" method.
>
> On Wed, Dec 7, 2022 at 6:14 AM Noel OConnor wrote:
>>
>> Hi,
>> I'm using a kafka source to re
27+Sources
[2]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-217%3A+Support+watermark+alignment+of+source+splits
On Thu, Dec 8, 2022 at 6:21 PM Niklas Wilcke
wrote:
> Hi Flink Community,
>
> I have a few questions regarding the new KafkaSource and event time, which
> I wasn'
nce/display/FLINK/FLIP-217%3A+Support+watermark+alignment+of+source+splits
>
> On Thu, Dec 8, 2022 at 6:21 PM Niklas Wilcke <mailto:niklas.wil...@uniberg.com>> wrote:
>> Hi Flink Community,
>>
>> I have a few questions regarding the new KafkaSource and event time,
/#execution-checkpointing-checkpoints-after-tasks-finish-enabled
On Wed, Sep 15, 2021 at 11:26 AM Lars Skjærven wrote:
> Using KafkaSource builder with a job parallelism larger than the number of
> kafka partitions, the job is unable to checkpoint.
>
> With a job parallelism of 4, 3 of t
Sep 15, 2021 at 11:26 AM Lars Skjærven wrote:
>
>> Using KafkaSource builder with a job parallelism larger than the number
>> of kafka partitions, the job is unable to checkpoint.
>>
>> With a job parallelism of 4, 3 of the tasks are marked as FINISHED for
>> th
execution-checkpointing-checkpoints-after-tasks-finish-enabled
>>
>> On Wed, Sep 15, 2021 at 11:26 AM Lars Skjærven wrote:
>>
>>> Using KafkaSource builder with a job parallelism larger than the number
>>> of kafka partitions, the job is unable to checkpoint.
>
Hi all,
The problem you are seeing Lars is somewhat intended behaviour, unfortunately.
With the batch/stream unification every Kafka partition is treated
as kind of workload assignment. If one subtask receives a signal that there is
no workload anymore it goes into the FINISHED state.
As alread
Thanks for the feedback.
> May I ask why you have less partitions than the parallelism? I would be
happy to learn more about your use-case to better understand the
> motivation.
The use case is that topic A, contains just a few messages with product
metadata that rarely gets updated, while topic
：
> The KafkaSource, and KafkaSourceBuilder appear to prevent users from
> providing their own KafkaSubscriber. Am I overlooking something?
>
> In my case I have an external system that controls which topics we should
> be ingesting, and it can change over time. I need to add, and rem
suppose you want to read from different topics every now and then? Does
> the topic-pattern option [1] in Table API Kafka connector meet your needs?
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-master/docs/connectors/table/kafka/#topic-pattern
>
> Preston Price 于20
ant to read from different topics every now and then? Does
>> the topic-pattern option [1] in Table API Kafka connector meet your needs?
>>
>> [1]
>> https://ci.apache.org/projects/flink/flink-docs-master/docs/connectors/table/kafka/#topic-pattern
>>
>> Preston
>>>
>>> I suppose you want to read from different topics every now and then?
>>> Does the topic-pattern option [1] in Table API Kafka connector meet your
>>> needs?
>>>
>>> [1]
>>> https://ci.apache.org/projects/flink/flink-docs-master/do
;>>
>>> On Wed, Oct 13, 2021 at 8:51 PM Caizhi Weng
>>> wrote:
>>>
>>>> Hi!
>>>>
>>>> I suppose you want to read from different topics every now and then?
>>>> Does the topic-pattern option [1] in Table API Kafka connec
t;. My understanding is it would not pick up new
>>>> topics that match the pattern after the job starts.
>>>>
>>>>
>>>> On Wed, Oct 13, 2021 at 8:51 PM Caizhi Weng
>>>> wrote:
>>>>
>>>>> Hi!
>>>>>
&
gt;>> topics that match the pattern after the job starts.
>>>>>
>>>>>
>>>>> On Wed, Oct 13, 2021 at 8:51 PM Caizhi Weng
>>>>> wrote:
>>>>>
>>>>>> Hi!
>>>>>>
>>>>>> I suppose yo
Oct 13, 2021 at 8:51 PM Caizhi Weng <mailto:tsreape...@gmail.com>> wrote:
> Hi!
>
> I suppose you want to read from different topics every now and then? Does the
> topic-pattern option [1] in Table API Kafka connector meet your needs?
>
> [1]
> https://ci.apache.org/
at would
>>>>>> fit the state of all potential topics. Additionally the documentation you
>>>>>> linked seems to suggest that the regular expression is evaluated only
>>>>>> once
>>>>>> "when the job starts run
starts running". My understanding is it would not pick up new topics that
>> match the pattern after the job starts.
>>
>>
>> On Wed, Oct 13, 2021 at 8:51 PM Caizhi Weng > <mailto:tsreape...@gmail.com>> wrote:
>> Hi!
>>
>> I suppose you
gt; system, so there's no way to craft a single regular expression that
>>>>>>> would
>>>>>>> fit the state of all potential topics. Additionally the documentation
>>>>>>> you
>>>>>>> linked seems to suggest
I've seen a few questions recently from folks migrating from
FlinkKafkaConsumer to KafkaSource that make me suspect that something has
changed.
In FlinkKafkaConsumerBase we have this code which sets a source subtask to
idle if all of its partitions are empty when the subtask s
s data from a locally saved Parquet File, and a KafkaSource consuming
> events from a remote Kafka broker.
>
> I was wondering if instead of using a local Parquet file, whether it is
> possible to directly stream the file from a GCS bucket and construct a File
> Source out of it
gt; We have a requirement as follows:
> >
> > We want to stream events from 2 sources: Parquet files stored in a GCS
> Bucket, and a Kafka topic.
> > With the release of Hybrid Source in Flink 1.14, we were able to
> construct a Hybrid Source which produces events from two sources:
nts from 2 sources: Parquet files stored in a GCS
>> > Bucket, and a Kafka topic.
>> > With the release of Hybrid Source in Flink 1.14, we were able to construct
>> > a Hybrid Source which produces events from two sources: a FileSource which
>> > reads data from
We are migrating to KafkaSource
<https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java>
from FlinkKafkaConsumer
<https://github.com/apache/flink/blob/master/flink-connectors/flink-connec
Hi David,
yes that's intentionally [1] as it could lead to correctness issues and it
was inconsistently used across sources. Yes it should be documented.
For now I'd put it in the KafkaSource docs because I'm not sure in which
release notes it would fit best. In which release
Thanks, Arvid.
Can you clarify how the KafkaSource currently behaves in the situation
where it starts with fewer partitions than subtasks? Is that the case
described in FLIP-180 as case 1: "Static assignment + too few splits"? The
implementation described there (emit MAX_WATERMARK) sh
wrote:
> Thanks, Arvid.
>
> Can you clarify how the KafkaSource currently behaves in the situation
> where it starts with fewer partitions than subtasks? Is that the case
> described in FLIP-180 as case 1: "Static assignment + too few splits"? The
> implementation de
Hi Santosh,
It’s best to avoid cross-posting. Let’s keep the discussion to SO.
Best regards,
Niklas
> On 12. Feb 2022, at 16:39, santosh joshi wrote:
>
> We are migrating to KafkaSource from FlinkKafkaConsumer. We have disabled
> auto commit of offset and instead committing them
om :
https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#starting-offset
Before in old FlinkKafkaConsumer it was from committed offset (i.e :
setStartFromGroupOffsets()
method)
which match with this behaviour in new KafkaSource : : OffsetsInitializer.
commit
Hi Bastien,
Thanks for asking. I didn't find any call of setStartFromGroupOffsets() within
Flink in the master branch. Could you please point out the code that
committed offset is used as default?
W.r.t. the new KafkaSource, if OffsetsInitializer.committedOffsets()
is used, an exception wi
fsets() within
> Flink in the master branch. Could you please point out the code that
> committed offset is used as default?
>
> W.r.t. the new KafkaSource, if OffsetsInitializer.committedOffsets()
> is used, an exception will be thrown at runtime in case there is no
> committed off
Flink in the master branch. Could you please point out the code that
>> committed offset is used as default?
>>
>> W.r.t. the new KafkaSource, if OffsetsInitializer.committedOffsets()
>> is used, an exception will be thrown at runtime in case there is no
>> committed offset,
e for sure .committedOffsets - we have it by default in our custom
KafkaSource builder to be sure we do not read all the previous data
(earliest)
What bother me is just this change in starting offset default behavior from
FlinkKafkaConsumer to KafkaSource (this can lead to mistake)
In fact it happe
请问这个邮件咋退订？
Replied Message
| From | bastien dine |
| Date | 06/15/2022 17:50 |
| To | Martijn Visser |
| Cc | Jing Ge,
user |
| Subject | Re: New KafkaSource API : Change in default behavior regarding
starting offset |
Hello Martijn,
Thanks for the link to the release note
15/2022 17:50
> To Martijn Visser
> Cc Jing Ge ,
> user
> Subject Re: New KafkaSource API : Change in default behavior regarding
> starting offset
> Hello Martijn,
>
> Thanks for the link to the release note, especially :
> "When resuming f
Hi everyone,
I have a few questions about reading KafkaSource state using the State
Processor API. I have a simple Flink application which reads from a Kafka
topic then produces to a different topic. After running the Flink job and
stopping it with a savepoint, I then write a few more records to
The FlinkKafkaConsumer that will be deprecated has the method
"setCommitOffsetsOnCheckpoints(boolan)" method.
However, that functionality is not the new KafkaSource class.
How is this behavior / functionality configured in the new API?
-Marco A. Villalobos
Hi all,
*tl;dr: Would it be possible to make a KafkaSource that uses Kafka's built
in consumer assignment for Flink tasks?*
At the Wikimedia Foundation we are evaluating
<https://phabricator.wikimedia.org/T307944> whether we can use a Kafka
'stretch' cluster to simplify
`KafkaPartitionSplitSerializer`.
Best,
Hang
Charles Tan 于2023年5月24日周三 06:27写道：
> Hi everyone,
>
> I have a few questions about reading KafkaSource state using the State
> Processor API. I have a simple Flink application which reads from a Kafka
> topic then produces to a different topic. After runnin
Hi Marco,
>
https://nightlies.apache.org/flink/flink-docs-stable/docs/connectors/datastream/kafka/#additional-properties
In the new KafkaSource, you can configure it in your properties. You can
take a look at `KafkaSourceOptions#COMMIT_OFFSETS_ON_CHECKPOINT` for the
specific config, which
:
> Hi all,
>
> *tl;dr: Would it be possible to make a KafkaSource that uses Kafka's built
> in consumer assignment for Flink tasks?*
>
> At the Wikimedia Foundation we are evaluating
> <https://phabricator.wikimedia.org/T307944> whether we can use a Kafka
>
1 - 100 of 104 matches
Mail list logo
