You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Salva Alcántara <sa...@gmail.com> on 2024/06/07 04:35:42 UTC
Setting uid hash for non-legacy sinks
Hi,
I want to add the uid for my Kafka sink in such a way that I can still use
the existing savepoint. The problem I'm having is that I cannot set the uid
hash. If I try something like this:
```
output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
```
I get the following error:
```
Exception in thread "main" java.lang.UnsupportedOperationException: Cannot
set a custom UID hash on a non-legacy sink
at
org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
```
How can one set the operator id directly then for new (non-legacy) sinks?
Kind regards,
Salva
Re: Setting uid hash for non-legacy sinks
Posted by Salva Alcántara <sa...@gmail.com>.
♥️
Salva Alcántara reacted via Gmail
<https://www.google.com/gmail/about/?utm_source=gmail-in-product&utm_medium=et&utm_campaign=emojireactionemail#app>
On Wed, Jun 12, 2024 at 1:48 PM Gabor Somogyi <ga...@gmail.com>
wrote:
> Hey Salva,
>
> Good to hear your issue has been resolved one way or another!
> Thanks for confirming that this operator hash trick is working on V2 as
> well.
>
> G
>
>
> On Wed, Jun 12, 2024 at 5:20 AM Salva Alcántara <sa...@gmail.com>
> wrote:
>
>> Hey Gabor,
>>
>> I didn't finally need to keep compatibility with existing savepoints so I
>> restarted my jobs from scratch so to say.
>>
>> However, I have given it a try locally and by debugging the job graph I
>> can see the hashes manually set following your recipe (see the
>> userDefinedOperatorIDs) so it seems to work fine!
>>
>> Regards,
>>
>> Salva
>>
>> On Mon, Jun 10, 2024 at 9:49 AM Gabor Somogyi <ga...@gmail.com>
>> wrote:
>>
>>> YW, ping me back whether it works because it's a nifty feature.
>>>
>>> G
>>>
>>> On Mon, Jun 10, 2024 at 9:26 AM Salva Alcántara <sa...@gmail.com>
>>> wrote:
>>>
>>>> Thanks Gabor, I will give it a try!
>>>>
>>>> On Mon, Jun 10, 2024 at 12:01 AM Gabor Somogyi <
>>>> gabor.g.somogyi@gmail.com> wrote:
>>>>
>>>>> Now I see the intention and then you must have a V2 sink, right? Maybe
>>>>> you look for the following:
>>>>>
>>>>> final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead";
>>>>> final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10";
>>>>> final CustomSinkOperatorUidHashes operatorsUidHashes =
>>>>> CustomSinkOperatorUidHashes.builder()
>>>>> .setWriterUidHash(writerHash)
>>>>> .setCommitterUidHash(committerHash)
>>>>> .build();
>>>>> src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME);
>>>>>
>>>>> G
>>>>>
>>>>>
>>>>> On Sun, Jun 9, 2024 at 1:21 PM Zhanghao Chen <
>>>>> zhanghao.chen@outlook.com> wrote:
>>>>>
>>>>>> Hi Salva,
>>>>>>
>>>>>> The SinkV2 transformation will be translated to multiple operators at
>>>>>> the physical level. When setting a UID, Flink will automatically generate
>>>>>> UID for sub-operators by filling the configured UID in a pre-defined naming
>>>>>> template. The naming template is carefully maintained to ensure
>>>>>> cross-version state compatibility. However, this cannot be easily done when
>>>>>> setting the uidHash, that's why Flink currently does not support setting it
>>>>>> for non-legacy sinks.
>>>>>>
>>>>>> A possible solution is to convert uidHash back to the uid and apply
>>>>>> the same strategy for generating uids to compute the corresponding uidHash
>>>>>> for each suboperator. Maybe you can further investigate it and fire a JIRA
>>>>>> issue on it.
>>>>>>
>>>>>> Best,
>>>>>> Zhanghao Chen
>>>>>> ------------------------------
>>>>>> *From:* Salva Alcántara <sa...@gmail.com>
>>>>>> *Sent:* Sunday, June 9, 2024 14:49
>>>>>> *To:* Gabor Somogyi <ga...@gmail.com>
>>>>>> *Cc:* user <us...@flink.apache.org>
>>>>>> *Subject:* Re: Setting uid hash for non-legacy sinks
>>>>>>
>>>>>> Hi Gabor,
>>>>>>
>>>>>> Yeah, I know this, but what if you initially forgot and now you want
>>>>>> to add the uid "after the fact"?
>>>>>>
>>>>>> You need to get the operator/vertex id used by Flink for current
>>>>>> savepoints and somehow set this id for the sink.
>>>>>>
>>>>>> With the uid method you would need to "hack" the existing hash (get a
>>>>>> string which when hashed produces it). I guess this can be done since
>>>>>> murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
>>>>>> precisely for that. From the javadocs:
>>>>>>
>>>>>> *Important*: this should be used as a workaround or for
>>>>>> troubleshooting. The provided hash needs to be unique per transformation
>>>>>> and job. A use case for this is in migration between Flink versions or
>>>>>> changing the jobs in a way that changes the automatically generated hashes.
>>>>>> In this case, providing the previous hashes directly through this method
>>>>>> (e.g. obtained from old logs) can help to reestablish a lost mapping from
>>>>>> states to their target operator.
>>>>>>
>>>>>>
>>>>>> ...but as I said, it seems this method is not supported for new
>>>>>> (non-legacy) sinks...
>>>>>>
>>>>>> Regards,
>>>>>>
>>>>>> Salva
>>>>>>
>>>>>> On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <
>>>>>> gabor.g.somogyi@gmail.com> wrote:
>>>>>>
>>>>>> Hi Salva,
>>>>>>
>>>>>> Just wondering why not good to set the uid like this?
>>>>>> ```
>>>>>> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
>>>>>> ```
>>>>>>
>>>>>> From the mentioned UID Flink is going to make the hash which is
>>>>>> consistent from UID -> HASH transformation perspective.
>>>>>>
>>>>>> BR,
>>>>>> G
>>>>>>
>>>>>>
>>>>>>
>>>>>> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <
>>>>>> salcantaraphd@gmail.com> wrote:
>>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> I want to add the uid for my Kafka sink in such a way that I can
>>>>>> still use the existing savepoint. The problem I'm having is that I cannot
>>>>>> set the uid hash. If I try something like this:
>>>>>>
>>>>>> ```
>>>>>>
>>>>>> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
>>>>>> ```
>>>>>>
>>>>>> I get the following error:
>>>>>>
>>>>>> ```
>>>>>> Exception in thread "main" java.lang.UnsupportedOperationException:
>>>>>> Cannot set a custom UID hash on a non-legacy sink
>>>>>> at
>>>>>> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
>>>>>> ```
>>>>>>
>>>>>> How can one set the operator id directly then for new (non-legacy)
>>>>>> sinks?
>>>>>>
>>>>>> Kind regards,
>>>>>>
>>>>>> Salva
>>>>>>
>>>>>>
Re: Setting uid hash for non-legacy sinks
Posted by Gabor Somogyi <ga...@gmail.com>.
Hey Salva,
Good to hear your issue has been resolved one way or another!
Thanks for confirming that this operator hash trick is working on V2 as
well.
G
On Wed, Jun 12, 2024 at 5:20 AM Salva Alcántara <sa...@gmail.com>
wrote:
> Hey Gabor,
>
> I didn't finally need to keep compatibility with existing savepoints so I
> restarted my jobs from scratch so to say.
>
> However, I have given it a try locally and by debugging the job graph I
> can see the hashes manually set following your recipe (see the
> userDefinedOperatorIDs) so it seems to work fine!
>
> Regards,
>
> Salva
>
> On Mon, Jun 10, 2024 at 9:49 AM Gabor Somogyi <ga...@gmail.com>
> wrote:
>
>> YW, ping me back whether it works because it's a nifty feature.
>>
>> G
>>
>> On Mon, Jun 10, 2024 at 9:26 AM Salva Alcántara <sa...@gmail.com>
>> wrote:
>>
>>> Thanks Gabor, I will give it a try!
>>>
>>> On Mon, Jun 10, 2024 at 12:01 AM Gabor Somogyi <
>>> gabor.g.somogyi@gmail.com> wrote:
>>>
>>>> Now I see the intention and then you must have a V2 sink, right? Maybe
>>>> you look for the following:
>>>>
>>>> final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead";
>>>> final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10";
>>>> final CustomSinkOperatorUidHashes operatorsUidHashes =
>>>> CustomSinkOperatorUidHashes.builder()
>>>> .setWriterUidHash(writerHash)
>>>> .setCommitterUidHash(committerHash)
>>>> .build();
>>>> src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME);
>>>>
>>>> G
>>>>
>>>>
>>>> On Sun, Jun 9, 2024 at 1:21 PM Zhanghao Chen <zh...@outlook.com>
>>>> wrote:
>>>>
>>>>> Hi Salva,
>>>>>
>>>>> The SinkV2 transformation will be translated to multiple operators at
>>>>> the physical level. When setting a UID, Flink will automatically generate
>>>>> UID for sub-operators by filling the configured UID in a pre-defined naming
>>>>> template. The naming template is carefully maintained to ensure
>>>>> cross-version state compatibility. However, this cannot be easily done when
>>>>> setting the uidHash, that's why Flink currently does not support setting it
>>>>> for non-legacy sinks.
>>>>>
>>>>> A possible solution is to convert uidHash back to the uid and apply
>>>>> the same strategy for generating uids to compute the corresponding uidHash
>>>>> for each suboperator. Maybe you can further investigate it and fire a JIRA
>>>>> issue on it.
>>>>>
>>>>> Best,
>>>>> Zhanghao Chen
>>>>> ------------------------------
>>>>> *From:* Salva Alcántara <sa...@gmail.com>
>>>>> *Sent:* Sunday, June 9, 2024 14:49
>>>>> *To:* Gabor Somogyi <ga...@gmail.com>
>>>>> *Cc:* user <us...@flink.apache.org>
>>>>> *Subject:* Re: Setting uid hash for non-legacy sinks
>>>>>
>>>>> Hi Gabor,
>>>>>
>>>>> Yeah, I know this, but what if you initially forgot and now you want
>>>>> to add the uid "after the fact"?
>>>>>
>>>>> You need to get the operator/vertex id used by Flink for current
>>>>> savepoints and somehow set this id for the sink.
>>>>>
>>>>> With the uid method you would need to "hack" the existing hash (get a
>>>>> string which when hashed produces it). I guess this can be done since
>>>>> murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
>>>>> precisely for that. From the javadocs:
>>>>>
>>>>> *Important*: this should be used as a workaround or for
>>>>> troubleshooting. The provided hash needs to be unique per transformation
>>>>> and job. A use case for this is in migration between Flink versions or
>>>>> changing the jobs in a way that changes the automatically generated hashes.
>>>>> In this case, providing the previous hashes directly through this method
>>>>> (e.g. obtained from old logs) can help to reestablish a lost mapping from
>>>>> states to their target operator.
>>>>>
>>>>>
>>>>> ...but as I said, it seems this method is not supported for new
>>>>> (non-legacy) sinks...
>>>>>
>>>>> Regards,
>>>>>
>>>>> Salva
>>>>>
>>>>> On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <
>>>>> gabor.g.somogyi@gmail.com> wrote:
>>>>>
>>>>> Hi Salva,
>>>>>
>>>>> Just wondering why not good to set the uid like this?
>>>>> ```
>>>>> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
>>>>> ```
>>>>>
>>>>> From the mentioned UID Flink is going to make the hash which is
>>>>> consistent from UID -> HASH transformation perspective.
>>>>>
>>>>> BR,
>>>>> G
>>>>>
>>>>>
>>>>>
>>>>> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <
>>>>> salcantaraphd@gmail.com> wrote:
>>>>>
>>>>> Hi,
>>>>>
>>>>> I want to add the uid for my Kafka sink in such a way that I can still
>>>>> use the existing savepoint. The problem I'm having is that I cannot set the
>>>>> uid hash. If I try something like this:
>>>>>
>>>>> ```
>>>>>
>>>>> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
>>>>> ```
>>>>>
>>>>> I get the following error:
>>>>>
>>>>> ```
>>>>> Exception in thread "main" java.lang.UnsupportedOperationException:
>>>>> Cannot set a custom UID hash on a non-legacy sink
>>>>> at
>>>>> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
>>>>> ```
>>>>>
>>>>> How can one set the operator id directly then for new (non-legacy)
>>>>> sinks?
>>>>>
>>>>> Kind regards,
>>>>>
>>>>> Salva
>>>>>
>>>>>
Re: Setting uid hash for non-legacy sinks
Posted by Salva Alcántara <sa...@gmail.com>.
Hey Gabor,
I didn't finally need to keep compatibility with existing savepoints so I
restarted my jobs from scratch so to say.
However, I have given it a try locally and by debugging the job graph I can
see the hashes manually set following your recipe (see the
userDefinedOperatorIDs) so it seems to work fine!
Regards,
Salva
On Mon, Jun 10, 2024 at 9:49 AM Gabor Somogyi <ga...@gmail.com>
wrote:
> YW, ping me back whether it works because it's a nifty feature.
>
> G
>
> On Mon, Jun 10, 2024 at 9:26 AM Salva Alcántara <sa...@gmail.com>
> wrote:
>
>> Thanks Gabor, I will give it a try!
>>
>> On Mon, Jun 10, 2024 at 12:01 AM Gabor Somogyi <ga...@gmail.com>
>> wrote:
>>
>>> Now I see the intention and then you must have a V2 sink, right? Maybe
>>> you look for the following:
>>>
>>> final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead";
>>> final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10";
>>> final CustomSinkOperatorUidHashes operatorsUidHashes =
>>> CustomSinkOperatorUidHashes.builder()
>>> .setWriterUidHash(writerHash)
>>> .setCommitterUidHash(committerHash)
>>> .build();
>>> src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME);
>>>
>>> G
>>>
>>>
>>> On Sun, Jun 9, 2024 at 1:21 PM Zhanghao Chen <zh...@outlook.com>
>>> wrote:
>>>
>>>> Hi Salva,
>>>>
>>>> The SinkV2 transformation will be translated to multiple operators at
>>>> the physical level. When setting a UID, Flink will automatically generate
>>>> UID for sub-operators by filling the configured UID in a pre-defined naming
>>>> template. The naming template is carefully maintained to ensure
>>>> cross-version state compatibility. However, this cannot be easily done when
>>>> setting the uidHash, that's why Flink currently does not support setting it
>>>> for non-legacy sinks.
>>>>
>>>> A possible solution is to convert uidHash back to the uid and apply the
>>>> same strategy for generating uids to compute the corresponding uidHash for
>>>> each suboperator. Maybe you can further investigate it and fire a JIRA
>>>> issue on it.
>>>>
>>>> Best,
>>>> Zhanghao Chen
>>>> ------------------------------
>>>> *From:* Salva Alcántara <sa...@gmail.com>
>>>> *Sent:* Sunday, June 9, 2024 14:49
>>>> *To:* Gabor Somogyi <ga...@gmail.com>
>>>> *Cc:* user <us...@flink.apache.org>
>>>> *Subject:* Re: Setting uid hash for non-legacy sinks
>>>>
>>>> Hi Gabor,
>>>>
>>>> Yeah, I know this, but what if you initially forgot and now you want to
>>>> add the uid "after the fact"?
>>>>
>>>> You need to get the operator/vertex id used by Flink for current
>>>> savepoints and somehow set this id for the sink.
>>>>
>>>> With the uid method you would need to "hack" the existing hash (get a
>>>> string which when hashed produces it). I guess this can be done since
>>>> murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
>>>> precisely for that. From the javadocs:
>>>>
>>>> *Important*: this should be used as a workaround or for
>>>> troubleshooting. The provided hash needs to be unique per transformation
>>>> and job. A use case for this is in migration between Flink versions or
>>>> changing the jobs in a way that changes the automatically generated hashes.
>>>> In this case, providing the previous hashes directly through this method
>>>> (e.g. obtained from old logs) can help to reestablish a lost mapping from
>>>> states to their target operator.
>>>>
>>>>
>>>> ...but as I said, it seems this method is not supported for new
>>>> (non-legacy) sinks...
>>>>
>>>> Regards,
>>>>
>>>> Salva
>>>>
>>>> On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <
>>>> gabor.g.somogyi@gmail.com> wrote:
>>>>
>>>> Hi Salva,
>>>>
>>>> Just wondering why not good to set the uid like this?
>>>> ```
>>>> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
>>>> ```
>>>>
>>>> From the mentioned UID Flink is going to make the hash which is
>>>> consistent from UID -> HASH transformation perspective.
>>>>
>>>> BR,
>>>> G
>>>>
>>>>
>>>>
>>>> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>
>>>> wrote:
>>>>
>>>> Hi,
>>>>
>>>> I want to add the uid for my Kafka sink in such a way that I can still
>>>> use the existing savepoint. The problem I'm having is that I cannot set the
>>>> uid hash. If I try something like this:
>>>>
>>>> ```
>>>>
>>>> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
>>>> ```
>>>>
>>>> I get the following error:
>>>>
>>>> ```
>>>> Exception in thread "main" java.lang.UnsupportedOperationException:
>>>> Cannot set a custom UID hash on a non-legacy sink
>>>> at
>>>> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
>>>> ```
>>>>
>>>> How can one set the operator id directly then for new (non-legacy)
>>>> sinks?
>>>>
>>>> Kind regards,
>>>>
>>>> Salva
>>>>
>>>>
Re: Setting uid hash for non-legacy sinks
Posted by Gabor Somogyi <ga...@gmail.com>.
YW, ping me back whether it works because it's a nifty feature.
G
On Mon, Jun 10, 2024 at 9:26 AM Salva Alcántara <sa...@gmail.com>
wrote:
> Thanks Gabor, I will give it a try!
>
> On Mon, Jun 10, 2024 at 12:01 AM Gabor Somogyi <ga...@gmail.com>
> wrote:
>
>> Now I see the intention and then you must have a V2 sink, right? Maybe
>> you look for the following:
>>
>> final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead";
>> final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10";
>> final CustomSinkOperatorUidHashes operatorsUidHashes =
>> CustomSinkOperatorUidHashes.builder()
>> .setWriterUidHash(writerHash)
>> .setCommitterUidHash(committerHash)
>> .build();
>> src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME);
>>
>> G
>>
>>
>> On Sun, Jun 9, 2024 at 1:21 PM Zhanghao Chen <zh...@outlook.com>
>> wrote:
>>
>>> Hi Salva,
>>>
>>> The SinkV2 transformation will be translated to multiple operators at
>>> the physical level. When setting a UID, Flink will automatically generate
>>> UID for sub-operators by filling the configured UID in a pre-defined naming
>>> template. The naming template is carefully maintained to ensure
>>> cross-version state compatibility. However, this cannot be easily done when
>>> setting the uidHash, that's why Flink currently does not support setting it
>>> for non-legacy sinks.
>>>
>>> A possible solution is to convert uidHash back to the uid and apply the
>>> same strategy for generating uids to compute the corresponding uidHash for
>>> each suboperator. Maybe you can further investigate it and fire a JIRA
>>> issue on it.
>>>
>>> Best,
>>> Zhanghao Chen
>>> ------------------------------
>>> *From:* Salva Alcántara <sa...@gmail.com>
>>> *Sent:* Sunday, June 9, 2024 14:49
>>> *To:* Gabor Somogyi <ga...@gmail.com>
>>> *Cc:* user <us...@flink.apache.org>
>>> *Subject:* Re: Setting uid hash for non-legacy sinks
>>>
>>> Hi Gabor,
>>>
>>> Yeah, I know this, but what if you initially forgot and now you want to
>>> add the uid "after the fact"?
>>>
>>> You need to get the operator/vertex id used by Flink for current
>>> savepoints and somehow set this id for the sink.
>>>
>>> With the uid method you would need to "hack" the existing hash (get a
>>> string which when hashed produces it). I guess this can be done since
>>> murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
>>> precisely for that. From the javadocs:
>>>
>>> *Important*: this should be used as a workaround or for
>>> troubleshooting. The provided hash needs to be unique per transformation
>>> and job. A use case for this is in migration between Flink versions or
>>> changing the jobs in a way that changes the automatically generated hashes.
>>> In this case, providing the previous hashes directly through this method
>>> (e.g. obtained from old logs) can help to reestablish a lost mapping from
>>> states to their target operator.
>>>
>>>
>>> ...but as I said, it seems this method is not supported for new
>>> (non-legacy) sinks...
>>>
>>> Regards,
>>>
>>> Salva
>>>
>>> On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <ga...@gmail.com>
>>> wrote:
>>>
>>> Hi Salva,
>>>
>>> Just wondering why not good to set the uid like this?
>>> ```
>>> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
>>> ```
>>>
>>> From the mentioned UID Flink is going to make the hash which is
>>> consistent from UID -> HASH transformation perspective.
>>>
>>> BR,
>>> G
>>>
>>>
>>>
>>> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>
>>> wrote:
>>>
>>> Hi,
>>>
>>> I want to add the uid for my Kafka sink in such a way that I can still
>>> use the existing savepoint. The problem I'm having is that I cannot set the
>>> uid hash. If I try something like this:
>>>
>>> ```
>>> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
>>> ```
>>>
>>> I get the following error:
>>>
>>> ```
>>> Exception in thread "main" java.lang.UnsupportedOperationException:
>>> Cannot set a custom UID hash on a non-legacy sink
>>> at
>>> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
>>> ```
>>>
>>> How can one set the operator id directly then for new (non-legacy) sinks?
>>>
>>> Kind regards,
>>>
>>> Salva
>>>
>>>
Re: Setting uid hash for non-legacy sinks
Posted by Salva Alcántara <sa...@gmail.com>.
Thanks Gabor, I will give it a try!
On Mon, Jun 10, 2024 at 12:01 AM Gabor Somogyi <ga...@gmail.com>
wrote:
> Now I see the intention and then you must have a V2 sink, right? Maybe you
> look for the following:
>
> final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead";
> final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10";
> final CustomSinkOperatorUidHashes operatorsUidHashes =
> CustomSinkOperatorUidHashes.builder()
> .setWriterUidHash(writerHash)
> .setCommitterUidHash(committerHash)
> .build();
> src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME);
>
> G
>
>
> On Sun, Jun 9, 2024 at 1:21 PM Zhanghao Chen <zh...@outlook.com>
> wrote:
>
>> Hi Salva,
>>
>> The SinkV2 transformation will be translated to multiple operators at the
>> physical level. When setting a UID, Flink will automatically generate UID
>> for sub-operators by filling the configured UID in a pre-defined naming
>> template. The naming template is carefully maintained to ensure
>> cross-version state compatibility. However, this cannot be easily done when
>> setting the uidHash, that's why Flink currently does not support setting it
>> for non-legacy sinks.
>>
>> A possible solution is to convert uidHash back to the uid and apply the
>> same strategy for generating uids to compute the corresponding uidHash for
>> each suboperator. Maybe you can further investigate it and fire a JIRA
>> issue on it.
>>
>> Best,
>> Zhanghao Chen
>> ------------------------------
>> *From:* Salva Alcántara <sa...@gmail.com>
>> *Sent:* Sunday, June 9, 2024 14:49
>> *To:* Gabor Somogyi <ga...@gmail.com>
>> *Cc:* user <us...@flink.apache.org>
>> *Subject:* Re: Setting uid hash for non-legacy sinks
>>
>> Hi Gabor,
>>
>> Yeah, I know this, but what if you initially forgot and now you want to
>> add the uid "after the fact"?
>>
>> You need to get the operator/vertex id used by Flink for current
>> savepoints and somehow set this id for the sink.
>>
>> With the uid method you would need to "hack" the existing hash (get a
>> string which when hashed produces it). I guess this can be done since
>> murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
>> precisely for that. From the javadocs:
>>
>> *Important*: this should be used as a workaround or for troubleshooting.
>> The provided hash needs to be unique per transformation and job. A use case
>> for this is in migration between Flink versions or changing the jobs in a
>> way that changes the automatically generated hashes. In this case,
>> providing the previous hashes directly through this method (e.g. obtained
>> from old logs) can help to reestablish a lost mapping from states to their
>> target operator.
>>
>>
>> ...but as I said, it seems this method is not supported for new
>> (non-legacy) sinks...
>>
>> Regards,
>>
>> Salva
>>
>> On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <ga...@gmail.com>
>> wrote:
>>
>> Hi Salva,
>>
>> Just wondering why not good to set the uid like this?
>> ```
>> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
>> ```
>>
>> From the mentioned UID Flink is going to make the hash which is
>> consistent from UID -> HASH transformation perspective.
>>
>> BR,
>> G
>>
>>
>>
>> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>
>> wrote:
>>
>> Hi,
>>
>> I want to add the uid for my Kafka sink in such a way that I can still
>> use the existing savepoint. The problem I'm having is that I cannot set the
>> uid hash. If I try something like this:
>>
>> ```
>> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
>> ```
>>
>> I get the following error:
>>
>> ```
>> Exception in thread "main" java.lang.UnsupportedOperationException:
>> Cannot set a custom UID hash on a non-legacy sink
>> at
>> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
>> ```
>>
>> How can one set the operator id directly then for new (non-legacy) sinks?
>>
>> Kind regards,
>>
>> Salva
>>
>>
Re: Setting uid hash for non-legacy sinks
Posted by Gabor Somogyi <ga...@gmail.com>.
Now I see the intention and then you must have a V2 sink, right? Maybe you
look for the following:
final String writerHash = "f6b178ce445dc3ffaa06bad27a51fead";
final String committerHash = "68ac8ae79eae4e3135a54f9689c4aa10";
final CustomSinkOperatorUidHashes operatorsUidHashes =
CustomSinkOperatorUidHashes.builder()
.setWriterUidHash(writerHash)
.setCommitterUidHash(committerHash)
.build();
src.sinkTo(sinkWithCommitter(), operatorsUidHashes).name(NAME);
G
On Sun, Jun 9, 2024 at 1:21 PM Zhanghao Chen <zh...@outlook.com>
wrote:
> Hi Salva,
>
> The SinkV2 transformation will be translated to multiple operators at the
> physical level. When setting a UID, Flink will automatically generate UID
> for sub-operators by filling the configured UID in a pre-defined naming
> template. The naming template is carefully maintained to ensure
> cross-version state compatibility. However, this cannot be easily done when
> setting the uidHash, that's why Flink currently does not support setting it
> for non-legacy sinks.
>
> A possible solution is to convert uidHash back to the uid and apply the
> same strategy for generating uids to compute the corresponding uidHash for
> each suboperator. Maybe you can further investigate it and fire a JIRA
> issue on it.
>
> Best,
> Zhanghao Chen
> ------------------------------
> *From:* Salva Alcántara <sa...@gmail.com>
> *Sent:* Sunday, June 9, 2024 14:49
> *To:* Gabor Somogyi <ga...@gmail.com>
> *Cc:* user <us...@flink.apache.org>
> *Subject:* Re: Setting uid hash for non-legacy sinks
>
> Hi Gabor,
>
> Yeah, I know this, but what if you initially forgot and now you want to
> add the uid "after the fact"?
>
> You need to get the operator/vertex id used by Flink for current
> savepoints and somehow set this id for the sink.
>
> With the uid method you would need to "hack" the existing hash (get a
> string which when hashed produces it). I guess this can be done since
> murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
> precisely for that. From the javadocs:
>
> *Important*: this should be used as a workaround or for troubleshooting.
> The provided hash needs to be unique per transformation and job. A use case
> for this is in migration between Flink versions or changing the jobs in a
> way that changes the automatically generated hashes. In this case,
> providing the previous hashes directly through this method (e.g. obtained
> from old logs) can help to reestablish a lost mapping from states to their
> target operator.
>
>
> ...but as I said, it seems this method is not supported for new
> (non-legacy) sinks...
>
> Regards,
>
> Salva
>
> On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <ga...@gmail.com>
> wrote:
>
> Hi Salva,
>
> Just wondering why not good to set the uid like this?
> ```
> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
> ```
>
> From the mentioned UID Flink is going to make the hash which is consistent
> from UID -> HASH transformation perspective.
>
> BR,
> G
>
>
>
> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>
> wrote:
>
> Hi,
>
> I want to add the uid for my Kafka sink in such a way that I can still use
> the existing savepoint. The problem I'm having is that I cannot set the uid
> hash. If I try something like this:
>
> ```
> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
> ```
>
> I get the following error:
>
> ```
> Exception in thread "main" java.lang.UnsupportedOperationException: Cannot
> set a custom UID hash on a non-legacy sink
> at
> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
> ```
>
> How can one set the operator id directly then for new (non-legacy) sinks?
>
> Kind regards,
>
> Salva
>
>
Re: Setting uid hash for non-legacy sinks
Posted by Zhanghao Chen <zh...@outlook.com>.
Hi Salva,
The SinkV2 transformation will be translated to multiple operators at the physical level. When setting a UID, Flink will automatically generate UID for sub-operators by filling the configured UID in a pre-defined naming template. The naming template is carefully maintained to ensure cross-version state compatibility. However, this cannot be easily done when setting the uidHash, that's why Flink currently does not support setting it for non-legacy sinks.
A possible solution is to convert uidHash back to the uid and apply the same strategy for generating uids to compute the corresponding uidHash for each suboperator. Maybe you can further investigate it and fire a JIRA issue on it.
Best,
Zhanghao Chen
________________________________
From: Salva Alcántara <sa...@gmail.com>
Sent: Sunday, June 9, 2024 14:49
To: Gabor Somogyi <ga...@gmail.com>
Cc: user <us...@flink.apache.org>
Subject: Re: Setting uid hash for non-legacy sinks
Hi Gabor,
Yeah, I know this, but what if you initially forgot and now you want to add the uid "after the fact"?
You need to get the operator/vertex id used by Flink for current savepoints and somehow set this id for the sink.
With the uid method you would need to "hack" the existing hash (get a string which when hashed produces it). I guess this can be done since murmur3 is a non-cryptographic hash but Flink has the "setUidHash" precisely for that. From the javadocs:
Important: this should be used as a workaround or for troubleshooting. The provided hash needs to be unique per transformation and job. A use case for this is in migration between Flink versions or changing the jobs in a way that changes the automatically generated hashes. In this case, providing the previous hashes directly through this method (e.g. obtained from old logs) can help to reestablish a lost mapping from states to their target operator.
...but as I said, it seems this method is not supported for new (non-legacy) sinks...
Regards,
Salva
On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <ga...@gmail.com>> wrote:
Hi Salva,
Just wondering why not good to set the uid like this?
```
output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
```
From the mentioned UID Flink is going to make the hash which is consistent from UID -> HASH transformation perspective.
BR,
G
On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>> wrote:
Hi,
I want to add the uid for my Kafka sink in such a way that I can still use the existing savepoint. The problem I'm having is that I cannot set the uid hash. If I try something like this:
```
output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
```
I get the following error:
```
Exception in thread "main" java.lang.UnsupportedOperationException: Cannot set a custom UID hash on a non-legacy sink
at org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
```
How can one set the operator id directly then for new (non-legacy) sinks?
Kind regards,
Salva
Re: Setting uid hash for non-legacy sinks
Posted by Salva Alcántara <sa...@gmail.com>.
Hi Gabor,
Yeah, I know this, but what if you initially forgot and now you want to add
the uid "after the fact"?
You need to get the operator/vertex id used by Flink for current savepoints
and somehow set this id for the sink.
With the uid method you would need to "hack" the existing hash (get a
string which when hashed produces it). I guess this can be done since
murmur3 is a non-cryptographic hash but Flink has the "setUidHash"
precisely for that. From the javadocs:
*Important*: this should be used as a workaround or for troubleshooting.
> The provided hash needs to be unique per transformation and job. A use case
> for this is in migration between Flink versions or changing the jobs in a
> way that changes the automatically generated hashes. In this case,
> providing the previous hashes directly through this method (e.g. obtained
> from old logs) can help to reestablish a lost mapping from states to their
> target operator.
...but as I said, it seems this method is not supported for new
(non-legacy) sinks...
Regards,
Salva
On Fri, Jun 7, 2024 at 10:20 AM Gabor Somogyi <ga...@gmail.com>
wrote:
> Hi Salva,
>
> Just wondering why not good to set the uid like this?
> ```
> output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
> ```
>
> From the mentioned UID Flink is going to make the hash which is consistent
> from UID -> HASH transformation perspective.
>
> BR,
> G
>
>
>
> On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>
> wrote:
>
>> Hi,
>>
>> I want to add the uid for my Kafka sink in such a way that I can still
>> use the existing savepoint. The problem I'm having is that I cannot set the
>> uid hash. If I try something like this:
>>
>> ```
>> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
>> ```
>>
>> I get the following error:
>>
>> ```
>> Exception in thread "main" java.lang.UnsupportedOperationException:
>> Cannot set a custom UID hash on a non-legacy sink
>> at
>> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
>> ```
>>
>> How can one set the operator id directly then for new (non-legacy) sinks?
>>
>> Kind regards,
>>
>> Salva
>>
>>
Re: Setting uid hash for non-legacy sinks
Posted by Gabor Somogyi <ga...@gmail.com>.
Hi Salva,
Just wondering why not good to set the uid like this?
```
output.sinkTo(outputSink).uid("my-human-readable-sink-uid");
```
From the mentioned UID Flink is going to make the hash which is consistent
from UID -> HASH transformation perspective.
BR,
G
On Fri, Jun 7, 2024 at 7:54 AM Salva Alcántara <sa...@gmail.com>
wrote:
> Hi,
>
> I want to add the uid for my Kafka sink in such a way that I can still use
> the existing savepoint. The problem I'm having is that I cannot set the uid
> hash. If I try something like this:
>
> ```
> output.sinkTo(outputSink).setUidHash("xyzbbbbbbbbbbbbbbbbbbbbbbbbbbbbb");
> ```
>
> I get the following error:
>
> ```
> Exception in thread "main" java.lang.UnsupportedOperationException: Cannot
> set a custom UID hash on a non-legacy sink
> at
> org.apache.flink.streaming.api.datastream.DataStreamSink.setUidHash(DataStreamSink.java:163)
> ```
>
> How can one set the operator id directly then for new (non-legacy) sinks?
>
> Kind regards,
>
> Salva
>
>