You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@flink.apache.org by Debraj Manna <su...@gmail.com> on 2021/06/23 13:56:09 UTC

"Legacy Source Thread" line in logs

Hi

I am seeing the below logs in flink 1.13.0 running in YARN

2021-06-23T13:41:45.761Z WARN grid.task.MetricSdmStalenessUtils Legacy
Source Thread - Source: MetricSource -> Filter -> MetricStoreMapper ->
(Filter -> Timestamps/Watermarks -> Map -> Flat Map, Sink:
FlinkKafkaProducer11, Sink: TSDBSink14) (1/1)#0
updateMetricStalenessInHisto:32 Received a non-positive staleness = -194239
at 1624455705761

Can someone let me know what does the "Legacy Source Thread" denotes?

I saw the same question here
<http://deprecated-apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/quot-Legacy-Source-Thread-quot-line-in-logs-td33941.html#a44594>
in the deprecated mailing list with no answer. So starting a new email
thread here.

Re: "Legacy Source Thread" line in logs

Posted by Debraj Manna <su...@gmail.com>.

Thanks Fabian again for the clarification.

On Thu, Jun 24, 2021 at 8:16 PM Fabian Paul <fa...@data-artisans.com>
wrote:

> Hi Debraj,
>
> Sorry for the confusion the FlinkKafkaConsumer is the old source and the
> overhauled one you can find here [1].
> You would need to replace the FlinkKafkaConsumer with the KafkaSource to
> not see the message anymore.
>
> Best
> Fabian
>
>
> [1]
> https://github.com/apache/flink/blob/2bd8fab01d2aba99a5f690d051e817d10d2c6f24/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java#L75
>
>

Re: "Legacy Source Thread" line in logs

Posted by Fabian Paul <fa...@data-artisans.com>.

Hi Debraj,

Sorry for the confusion the FlinkKafkaConsumer is the old source and the overhauled one you can find here [1].
You would need to replace the FlinkKafkaConsumer with the KafkaSource to not see the message anymore.

Best
Fabian


[1] https://github.com/apache/flink/blob/2bd8fab01d2aba99a5f690d051e817d10d2c6f24/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSource.java#L75

Re: "Legacy Source Thread" line in logs

Posted by Debraj Manna <su...@gmail.com>.

Thanks Fabian for replying. But I am using KafkaSource only.

The code is something like below.

class MetricSource {
      final Set<String> metricSdms = new HashSet();
      ...
       env.addSource(MetricKafkaSourceFactory.createConsumer(jobParams))

        .name(MetricSource.class.getSimpleName())
        .uid(MetricSource.class.getSimpleName())
        .filter(sdm -> metricSdms.contains(sdm.getType()));

}

class MetricKafkaSourceFactory {
      public static FlinkKafkaConsumer<SelfDescribingMessageDO>
createConsumer(final Configuration jobParams) {
           ...
       return new FlinkKafkaConsumer<>(topic, new DeserializationSchema(),
props);
   }
}


On Wed, Jun 23, 2021 at 7:31 PM Fabian Paul <fa...@data-artisans.com>
wrote:

> Hi Debraj,
>
> By Source Legacy Thread we refer to all sources which do not implement the
> new interface yet [1]. Currently only the Hive, Kafka and FileSource
> are already migrated. In general, there is no sever downside of using the
> older source but in the future we plan only to implement ones based on
> the new operator model.
>
> Best,
> Fabian
>
> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface
>
>

Re: "Legacy Source Thread" line in logs

Posted by Fabian Paul <fa...@data-artisans.com>.

Hi Debraj,

By Source Legacy Thread we refer to all sources which do not implement the new interface yet [1]. Currently only the Hive, Kafka and FileSource
are already migrated. In general, there is no sever downside of using the older source but in the future we plan only to implement ones based on 
the new operator model.

Best,
Fabian

[1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface