I was glad to find that bravo had now been updated to support installing
bravo to a local maven repo.
I was able to load a checkpoint created by my job, thanks to the example
provided in bravo README, but I'm still missing the essential piece.
My code was:
OperatorStateReader reader = new OperatorStateReader(env2,
savepoint, "DistinctFunction");
DontKnowWhatTypeThisIs reducingState = reader.readKeyedStates(what
should I put here?);
I don't know how to read the values collected from reduce() calls in the
state. Is there a way to access the reducing state of the window with
bravo? I'm a bit confused how this works, because when I check with
debugger, flink internally uses a ReducingStateDescriptor
with name=window-contents, but still reading operator state for
"DistinctFunction" didn't at least throw an exception ("window-contents"
threw – obviously there's no operator by that name).
Cheers,
Juho
On Mon, Oct 15, 2018 at 2:25 PM Juho Autio <juho.au...@rovio.com> wrote:
> Hi Stefan,
>
> Sorry but it doesn't seem immediately clear to me what's a good way to use
> https://github.com/king/bravo.
>
> How are people using it? Would you for example modify build.gradle somehow
> to publish the bravo as a library locally/internally? Or add code directly
> in the bravo project (locally) and run it from there (using an IDE, for
> example)? Also it doesn't seem like the bravo gradle project supports
> building a flink job jar, but if it does, how do I do it?
>
> Thanks.
>
> On Thu, Oct 4, 2018 at 9:30 PM Juho Autio <juho.au...@rovio.com> wrote:
>
>> Good then, I'll try to analyze the savepoints with Bravo. Thanks!
>>
>> > How would you assume that backpressure would influence your updates?
>> Updates to each local state still happen event-by-event, in a single
>> reader/writing thread.
>>
>> Sure, just an ignorant guess by me. I'm not familiar with most of Flink's
>> internals. Any way high backpressure is not a seen on this job after it has
>> caught up the lag, so at I thought it would be worth mentioning.
>>
>> On Thu, Oct 4, 2018 at 6:24 PM Stefan Richter <
>> s.rich...@data-artisans.com> wrote:
>>
>>> Hi,
>>>
>>> Am 04.10.2018 um 16:08 schrieb Juho Autio <juho.au...@rovio.com>:
>>>
>>> > you could take a look at Bravo [1] to query your savepoints and to
>>> check if the state in the savepoint complete w.r.t your expectations
>>>
>>> Thanks. I'm not 100% if this is the case, but to me it seemed like the
>>> missed ids were being logged by the reducer soon after the job had started
>>> (after restoring a savepoint). But on the other hand, after that I also
>>> made another savepoint & restored that, so what I could check is: does that
>>> next savepoint have the missed ids that were logged (a couple of minutes
>>> before the savepoint was created, so there should've been more than enough
>>> time to add them to the state before the savepoint was triggered) or not.
>>> Any way, if I would be able to verify with Bravo that the ids are missing
>>> from the savepoint (even though reduced logged that it saw them), would
>>> that help in figuring out where they are lost? Is there some major
>>> difference compared to just looking at the final output after window has
>>> been triggered?
>>>
>>>
>>>
>>> I think that makes a difference. For example, you can investigate if
>>> there is a state loss or a problem with the windowing. In the savepoint you
>>> could see which keys exists and to which windows they are assigned. Also
>>> just to make sure there is no misunderstanding: only elements that are in
>>> the state at the start of a savepoint are expected to be part of the
>>> savepoint; all elements between start and completion of the savepoint are
>>> not expected to be part of the savepoint.
>>>
>>>
>>> > I also doubt that the problem is about backpressure after restore,
>>> because the job will only continue running after the state restore is
>>> already completed.
>>>
>>> Yes, I'm not suspecting that the state restoring would be the problem
>>> either. My concern was about backpressure possibly messing with the updates
>>> of reducing state? I would tend to suspect that updating the state
>>> consistently is what fails, where heavy load / backpressure might be a
>>> factor.
>>>
>>>
>>>
>>> How would you assume that backpressure would influence your updates?
>>> Updates to each local state still happen event-by-event, in a single
>>> reader/writing thread.
>>>
>>>
>>> On Thu, Oct 4, 2018 at 4:18 PM Stefan Richter <
>>> s.rich...@data-artisans.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> you could take a look at Bravo [1] to query your savepoints and to
>>>> check if the state in the savepoint complete w.r.t your expectations. I
>>>> somewhat doubt that there is a general problem with the state/savepoints
>>>> because many users are successfully running it on a large state and I am
>>>> not aware of any data loss problems, but nothing is impossible. What the
>>>> savepoint does is also straight forward: iterate a db snapshot and write
>>>> all key/value pairs to disk, so all data that was in the db at the time of
>>>> the savepoint, should show up. I also doubt that the problem is about
>>>> backpressure after restore, because the job will only continue running
>>>> after the state restore is already completed. Did you check if you are
>>>> using exactly-one-semantics or at-least-once semantics? Also did you check
>>>> that the kafka consumer start position is configured properly [2]? Are
>>>> watermarks generated as expected after restore?
>>>>
>>>> One more unrelated high-level comment that I have: for a granularity of
>>>> 24h windows, I wonder if it would not make sense to use a batch job
>>>> instead?
>>>>
>>>> Best,
>>>> Stefan
>>>>
>>>> [1] https://github.com/king/bravo
>>>> [2]
>>>> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#kafka-consumers-start-position-configuration
>>>>
>>>> Am 04.10.2018 um 14:53 schrieb Juho Autio <juho.au...@rovio.com>:
>>>>
>>>> Thanks for the suggestions!
>>>>
>>>> > In general, it would be tremendously helpful to have a minimal
>>>> working example which allows to reproduce the problem.
>>>>
>>>> Definitely. The problem with reproducing has been that this only seems
>>>> to happen in the bigger production data volumes.
>>>>
>>>> That's why I'm hoping to find a way to debug this with the production
>>>> data. With that it seems to consistently cause some misses every time the
>>>> job is killed/restored.
>>>>
>>>> > check if it happens for shorter windows, like 1h etc
>>>>
>>>> What would be the benefit of that compared to 24h window?
>>>>
>>>> > simplify the job to not use a reduce window but simply a time window
>>>> which outputs the window events. Then counting the input and output events
>>>> should allow you to verify the results. If you are not seeing missing
>>>> events, then it could have something to do with the reducing state used in
>>>> the reduce function.
>>>>
>>>> Hm, maybe, but not sure how useful that would be, because it wouldn't
>>>> yet prove that it's related to reducing, because not having a reduce
>>>> function could also mean smaller load on the job, which might alone be
>>>> enough to make the problem not manifest.
>>>>
>>>> Is there a way to debug what goes into the reducing state (including
>>>> what gets removed or overwritten and what restored), if that makes sense..?
>>>> Maybe some suitable logging could be used to prove that the lost data is
>>>> written to the reducing state (or at least asked to be written), but not
>>>> found any more when the window closes and state is flushed?
>>>>
>>>> On configuration once more, we're using RocksDB state backend with
>>>> asynchronous incremental checkpointing. The state is restored from
>>>> savepoints though, we haven't been using those checkpoints in these tests
>>>> (although they could be used in case of crashes – but we haven't had those
>>>> now).
>>>>
>>>> On Thu, Oct 4, 2018 at 3:25 PM Till Rohrmann <trohrm...@apache.org>
>>>> wrote:
>>>>
>>>>> Hi Juho,
>>>>>
>>>>> another idea to further narrow down the problem could be to simplify
>>>>> the job to not use a reduce window but simply a time window which outputs
>>>>> the window events. Then counting the input and output events should allow
>>>>> you to verify the results. If you are not seeing missing events, then it
>>>>> could have something to do with the reducing state used in the reduce
>>>>> function.
>>>>>
>>>>> In general, it would be tremendously helpful to have a minimal working
>>>>> example which allows to reproduce the problem.
>>>>>
>>>>> Cheers,
>>>>> Till
>>>>>
>>>>> On Thu, Oct 4, 2018 at 2:02 PM Andrey Zagrebin <
>>>>> and...@data-artisans.com> wrote:
>>>>>
>>>>>> Hi Juho,
>>>>>>
>>>>>> can you try to reduce the job to minimal reproducible example and
>>>>>> share the job and input?
>>>>>>
>>>>>> For example:
>>>>>> - some simple records as input, e.g. tuples of primitive types saved
>>>>>> as cvs
>>>>>> - minimal deduplication job which processes them and misses records
>>>>>> - check if it happens for shorter windows, like 1h etc
>>>>>> - setup which you use for the job, ideally locally reproducible or
>>>>>> cloud
>>>>>>
>>>>>> Best,
>>>>>> Andrey
>>>>>>
>>>>>> On 4 Oct 2018, at 11:13, Juho Autio <juho.au...@rovio.com> wrote:
>>>>>>
>>>>>> Sorry to insist, but we seem to be blocked for any serious usage of
>>>>>> state in Flink if we can't rely on it to not miss data in case of
>>>>>> restore.
>>>>>>
>>>>>> Would anyone have suggestions for how to troubleshoot this? So far I
>>>>>> have verified with DEBUG logs that our reduce function gets to process
>>>>>> also
>>>>>> the data that is missing from window output.
>>>>>>
>>>>>> On Mon, Oct 1, 2018 at 11:56 AM Juho Autio <juho.au...@rovio.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Hi Andrey,
>>>>>>>
>>>>>>> To rule out for good any questions about sink behaviour, the job was
>>>>>>> killed and started with an additional Kafka sink.
>>>>>>>
>>>>>>> The same number of ids were missed in both outputs: KafkaSink &
>>>>>>> BucketingSink.
>>>>>>>
>>>>>>> I wonder what would be the next steps in debugging?
>>>>>>>
>>>>>>> On Fri, Sep 21, 2018 at 3:49 PM Juho Autio <juho.au...@rovio.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Thanks, Andrey.
>>>>>>>>
>>>>>>>> > so it means that the savepoint does not loose at least some
>>>>>>>> dropped records.
>>>>>>>>
>>>>>>>> I'm not sure what you mean by that? I mean, it was known from the
>>>>>>>> beginning, that not everything is lost before/after restoring a
>>>>>>>> savepoint,
>>>>>>>> just some records around the time of restoration. It's not 100% clear
>>>>>>>> whether records are lost before making a savepoint or after restoring
>>>>>>>> it.
>>>>>>>> Although, based on the new DEBUG logs it seems more like losing some
>>>>>>>> records that are seen ~soon after restoring. It seems like Flink would
>>>>>>>> be
>>>>>>>> somehow confused either about the restored state vs. new inserts to
>>>>>>>> state.
>>>>>>>> This could also be somehow linked to the high back pressure on the
>>>>>>>> kafka
>>>>>>>> source while the stream is catching up.
>>>>>>>>
>>>>>>>> > If it is feasible for your setup, I suggest to insert one more
>>>>>>>> map function after reduce and before sink.
>>>>>>>> > etc.
>>>>>>>>
>>>>>>>> Isn't that the same thing that we discussed before? Nothing is sent
>>>>>>>> to BucketingSink before the window closes, so I don't see how it would
>>>>>>>> make
>>>>>>>> any difference if we replace the BucketingSink with a map function or
>>>>>>>> another sink type. We don't create or restore savepoints during the
>>>>>>>> time
>>>>>>>> when BucketingSink gets input or has open buckets – that happens at a
>>>>>>>> much
>>>>>>>> later time of day. I would focus on figuring out why the records are
>>>>>>>> lost
>>>>>>>> while the window is open. But I don't know how to do that. Would you
>>>>>>>> have
>>>>>>>> any additional suggestions?
>>>>>>>>
>>>>>>>> On Fri, Sep 21, 2018 at 3:30 PM Andrey Zagrebin <
>>>>>>>> and...@data-artisans.com> wrote:
>>>>>>>>
>>>>>>>>> Hi Juho,
>>>>>>>>>
>>>>>>>>> so it means that the savepoint does not loose at least some
>>>>>>>>> dropped records.
>>>>>>>>>
>>>>>>>>> If it is feasible for your setup, I suggest to insert one more map
>>>>>>>>> function after reduce and before sink.
>>>>>>>>> The map function should be called right after window is triggered
>>>>>>>>> but before flushing to s3.
>>>>>>>>> The result of reduce (deduped record) could be logged there.
>>>>>>>>> This should allow to check whether the processed distinct records
>>>>>>>>> were buffered in the state after the restoration from the savepoint
>>>>>>>>> or not.
>>>>>>>>> If they were buffered we should see that there was an attempt to
>>>>>>>>> write them
>>>>>>>>> to the sink from the state.
>>>>>>>>>
>>>>>>>>> Another suggestion is to try to write records to some other sink
>>>>>>>>> or to both.
>>>>>>>>> E.g. if you can access file system of workers, maybe just into
>>>>>>>>> local files and check whether the records are also dropped there.
>>>>>>>>>
>>>>>>>>> Best,
>>>>>>>>> Andrey
>>>>>>>>>
>>>>>>>>> On 20 Sep 2018, at 15:37, Juho Autio <juho.au...@rovio.com> wrote:
>>>>>>>>>
>>>>>>>>> Hi Andrey!
>>>>>>>>>
>>>>>>>>> I was finally able to gather the DEBUG logs that you suggested. In
>>>>>>>>> short, the reducer logged that it processed at least some of the ids
>>>>>>>>> that
>>>>>>>>> were missing from the output.
>>>>>>>>>
>>>>>>>>> "At least some", because I didn't have the job running with DEBUG
>>>>>>>>> logs for the full 24-hour window period. So I was only able to look
>>>>>>>>> up if I
>>>>>>>>> can find *some* of the missing ids in the DEBUG logs. Which I did
>>>>>>>>> indeed.
>>>>>>>>>
>>>>>>>>> I changed the DistinctFunction.java to do this:
>>>>>>>>>
>>>>>>>>> @Override
>>>>>>>>> public Map<String, String> reduce(Map<String, String> value1,
>>>>>>>>> Map<String, String> value2) {
>>>>>>>>> LOG.debug("DistinctFunction.reduce returns: {}={}",
>>>>>>>>> value1.get("field"), value1.get("id"));
>>>>>>>>> return value1;
>>>>>>>>> }
>>>>>>>>>
>>>>>>>>> Then:
>>>>>>>>>
>>>>>>>>> vi flink-1.6.0/conf/log4j.properties
>>>>>>>>>
>>>>>>>>> log4j.logger.org.apache.flink.streaming.runtime.tasks.StreamTask=DEBUG
>>>>>>>>> log4j.logger.com.rovio.ds.flink.uniqueid.DistinctFunction=DEBUG
>>>>>>>>>
>>>>>>>>> Then I ran the following kind of test:
>>>>>>>>>
>>>>>>>>> - Cancelled the on-going job with savepoint created at ~Sep 18
>>>>>>>>> 08:35 UTC 2018
>>>>>>>>> - Started a new cluster & job with DEBUG enabled at ~09:13,
>>>>>>>>> restored from that previous cluster's savepoint
>>>>>>>>> - Ran until caught up offsets
>>>>>>>>> - Cancelled the job with a new savepoint
>>>>>>>>> - Started a new job _without_ DEBUG, which restored the new
>>>>>>>>> savepoint, let it keep running so that it will eventually write the
>>>>>>>>> output
>>>>>>>>>
>>>>>>>>> Then on the next day, after results had been flushed when the
>>>>>>>>> 24-hour window closed, I compared the results again with a batch
>>>>>>>>> version's
>>>>>>>>> output. And found some missing ids as usual.
>>>>>>>>>
>>>>>>>>> I drilled down to one specific missing id (I'm replacing the
>>>>>>>>> actual value with AN12345 below), which was not found in the stream
>>>>>>>>> output,
>>>>>>>>> but was found in batch output & flink DEBUG logs.
>>>>>>>>>
>>>>>>>>> Related to that id, I gathered the following information:
>>>>>>>>>
>>>>>>>>> 2018-09-18~09:13:21,000 job started & savepoint is restored
>>>>>>>>>
>>>>>>>>> 2018-09-18 09:14:29,085 missing id is processed for the first
>>>>>>>>> time, proved by this log line:
>>>>>>>>> 2018-09-18 09:14:29,085 DEBUG
>>>>>>>>> com.rovio.ds.flink.uniqueid.DistinctFunction -
>>>>>>>>> DistinctFunction.reduce returns: s.aid1=AN12345
>>>>>>>>>
>>>>>>>>> 2018-09-18 09:15:14,264 first synchronous part of checkpoint
>>>>>>>>> 2018-09-18 09:15:16,544 first asynchronous part of checkpoint
>>>>>>>>>
>>>>>>>>> (
>>>>>>>>> more occurrences of checkpoints (~1 min checkpointing time + ~1
>>>>>>>>> min delay before next)
>>>>>>>>> /
>>>>>>>>> more occurrences of DistinctFunction.reduce
>>>>>>>>> )
>>>>>>>>>
>>>>>>>>> 2018-09-18 09:23:45,053 missing id is processed for the last time
>>>>>>>>>
>>>>>>>>> 2018-09-18~10:20:00,000 savepoint created & job cancelled
>>>>>>>>>
>>>>>>>>> To be noted, there was high backpressure after restoring from
>>>>>>>>> savepoint until the stream caught up with the kafka offsets.
>>>>>>>>> Although, our
>>>>>>>>> job uses assign timestamps & watermarks on the flink kafka consumer
>>>>>>>>> itself,
>>>>>>>>> so event time of all partitions is synchronized. As expected, we
>>>>>>>>> don't get
>>>>>>>>> any late data in the late data side output.
>>>>>>>>>
>>>>>>>>> From this we can see that the missing ids are processed by the
>>>>>>>>> reducer, but they must get lost somewhere before the 24-hour window is
>>>>>>>>> triggered.
>>>>>>>>>
>>>>>>>>> I think it's worth mentioning once more that the stream doesn't
>>>>>>>>> miss any ids if we let it's running without interruptions / state
>>>>>>>>> restoring.
>>>>>>>>>
>>>>>>>>> What's next?
>>>>>>>>>
>>>>>>>>> On Wed, Aug 29, 2018 at 3:49 PM Andrey Zagrebin <
>>>>>>>>> and...@data-artisans.com> wrote:
>>>>>>>>>
>>>>>>>>>> Hi Juho,
>>>>>>>>>>
>>>>>>>>>> > only when the 24-hour window triggers, BucketingSink gets a
>>>>>>>>>> burst of input
>>>>>>>>>>
>>>>>>>>>> This is of course totally true, my understanding is the same. We
>>>>>>>>>> cannot exclude problem there for sure, just savepoints are used a
>>>>>>>>>> lot w/o
>>>>>>>>>> problem reports and BucketingSink is known to be problematic with
>>>>>>>>>> s3. That
>>>>>>>>>> is why, I asked you:
>>>>>>>>>>
>>>>>>>>>> > You also wrote that the timestamps of lost event are 'probably'
>>>>>>>>>> around the time of the savepoint, if it is not yet for sure I would
>>>>>>>>>> also
>>>>>>>>>> check it.
>>>>>>>>>>
>>>>>>>>>> Although, bucketing sink might loose any data at the end of the
>>>>>>>>>> day (also from the middle). The fact, that it is always around the
>>>>>>>>>> time of
>>>>>>>>>> taking a savepoint and not random, is surely suspicious and possible
>>>>>>>>>> savepoint failures need to be investigated.
>>>>>>>>>>
>>>>>>>>>> Regarding the s3 problem, s3 doc says:
>>>>>>>>>>
>>>>>>>>>> > The caveat is that if you make a HEAD or GET request to the key
>>>>>>>>>> name (to find if the object exists) before creating the object,
>>>>>>>>>> Amazon S3
>>>>>>>>>> provides 'eventual consistency' for read-after-write.
>>>>>>>>>>
>>>>>>>>>> The algorithm you suggest is how it is roughly implemented now
>>>>>>>>>> (BucketingSink.openNewPartFile). My understanding is that
>>>>>>>>>> 'eventual consistency’ means that even if you just created file (its
>>>>>>>>>> name
>>>>>>>>>> is key) it can be that you do not get it in the list or exists (HEAD)
>>>>>>>>>> returns false and you risk to rewrite the previous part.
>>>>>>>>>>
>>>>>>>>>> The BucketingSink was designed for a standard file system. s3 is
>>>>>>>>>> used over a file system wrapper atm but does not always provide
>>>>>>>>>> normal file
>>>>>>>>>> system guarantees. See also last example in [1].
>>>>>>>>>>
>>>>>>>>>> Cheers,
>>>>>>>>>> Andrey
>>>>>>>>>>
>>>>>>>>>> [1]
>>>>>>>>>> https://codeburst.io/quick-explanation-of-the-s3-consistency-model-6c9f325e3f82
>>>>>>>>>>
>>>>>>>>>> On 29 Aug 2018, at 12:11, Juho Autio <juho.au...@rovio.com>
>>>>>>>>>> wrote:
>>>>>>>>>>
>>>>>>>>>> Andrey, thank you very much for the debugging suggestions, I'll
>>>>>>>>>> try them.
>>>>>>>>>>
>>>>>>>>>> In the meanwhile two more questions, please:
>>>>>>>>>>
>>>>>>>>>> > Just to keep in mind this problem with s3 and exclude it for
>>>>>>>>>> sure. I would also check whether the size of missing events is
>>>>>>>>>> around the
>>>>>>>>>> batch size of BucketingSink or not.
>>>>>>>>>>
>>>>>>>>>> Fair enough, but I also want to focus on debugging the most
>>>>>>>>>> probable subject first. So what do you think about this – true or
>>>>>>>>>> false:
>>>>>>>>>> only when the 24-hour window triggers, BucketinSink gets a burst of
>>>>>>>>>> input.
>>>>>>>>>> Around the state restoring point (middle of the day) it doesn't get
>>>>>>>>>> any
>>>>>>>>>> input, so it can't lose anything either. Isn't this true, or have I
>>>>>>>>>> totally
>>>>>>>>>> missed how Flink works in triggering window results? I would not
>>>>>>>>>> expect
>>>>>>>>>> there to be any optimization that speculatively triggers early
>>>>>>>>>> results of a
>>>>>>>>>> regular time window to the downstream operators.
>>>>>>>>>>
>>>>>>>>>> > The old BucketingSink has in general problem with s3.
>>>>>>>>>> Internally BucketingSink queries s3 as a file system to list already
>>>>>>>>>> written file parts (batches) and determine index of the next part to
>>>>>>>>>> start.
>>>>>>>>>> Due to eventual consistency of checking file existence in s3 [1], the
>>>>>>>>>> BucketingSink can rewrite the previously written part and basically
>>>>>>>>>> loose
>>>>>>>>>> it.
>>>>>>>>>>
>>>>>>>>>> I was wondering, what does S3's "read-after-write consistency"
>>>>>>>>>> (mentioned on the page you linked) actually mean. It seems that this
>>>>>>>>>> might
>>>>>>>>>> be possible:
>>>>>>>>>> - LIST keys, find current max index
>>>>>>>>>> - choose next index = max + 1
>>>>>>>>>> - HEAD next index: if it exists, keep adding + 1 until key
>>>>>>>>>> doesn't exist on S3
>>>>>>>>>>
>>>>>>>>>> But definitely sounds easier if a sink keeps track of files in a
>>>>>>>>>> way that's guaranteed to be consistent.
>>>>>>>>>>
>>>>>>>>>> Cheers,
>>>>>>>>>> Juho
>>>>>>>>>>
>>>>>>>>>> On Mon, Aug 27, 2018 at 2:04 PM Andrey Zagrebin <
>>>>>>>>>> and...@data-artisans.com> wrote:
>>>>>>>>>>
>>>>>>>>>>> Hi,
>>>>>>>>>>>
>>>>>>>>>>> true, StreamingFileSink does not support s3 in 1.6.0, it is
>>>>>>>>>>> planned for the next 1.7 release, sorry for confusion.
>>>>>>>>>>> The old BucketingSink has in general problem with s3.
>>>>>>>>>>> Internally BucketingSink queries s3 as a file system
>>>>>>>>>>> to list already written file parts (batches) and determine index
>>>>>>>>>>> of the next part to start. Due to eventual consistency of checking
>>>>>>>>>>> file
>>>>>>>>>>> existence in s3 [1], the BucketingSink can rewrite the previously
>>>>>>>>>>> written
>>>>>>>>>>> part and basically loose it. It should be fixed for
>>>>>>>>>>> StreamingFileSink in
>>>>>>>>>>> 1.7 where Flink keeps its own track of written parts and does not
>>>>>>>>>>> rely on
>>>>>>>>>>> s3 as a file system.
>>>>>>>>>>> I also include Kostas, he might add more details.
>>>>>>>>>>>
>>>>>>>>>>> Just to keep in mind this problem with s3 and exclude it for
>>>>>>>>>>> sure I would also check whether the size of missing events is
>>>>>>>>>>> around the
>>>>>>>>>>> batch size of BucketingSink or not. You also wrote that the
>>>>>>>>>>> timestamps of
>>>>>>>>>>> lost event are 'probably' around the time of the savepoint, if it
>>>>>>>>>>> is not
>>>>>>>>>>> yet for sure I would also check it.
>>>>>>>>>>>
>>>>>>>>>>> Have you already checked the log files of job manager and task
>>>>>>>>>>> managers for the job running before and after the restore from the
>>>>>>>>>>> check
>>>>>>>>>>> point? Is everything successful there, no errors, relevant warnings
>>>>>>>>>>> or
>>>>>>>>>>> exceptions?
>>>>>>>>>>>
>>>>>>>>>>> As the next step, I would suggest to log all encountered events
>>>>>>>>>>> in DistinctFunction.reduce if possible for production data and check
>>>>>>>>>>> whether the missed events are eventually processed before or after
>>>>>>>>>>> the
>>>>>>>>>>> savepoint. The following log message indicates a border between the
>>>>>>>>>>> events
>>>>>>>>>>> that should be included into the savepoint (logged before) or not:
>>>>>>>>>>> “{} ({}, synchronous part) in thread {} took {} ms” (template)
>>>>>>>>>>> Also check if the savepoint has been overall completed:
>>>>>>>>>>> "{} ({}, asynchronous part) in thread {} took {} ms."
>>>>>>>>>>>
>>>>>>>>>>> Best,
>>>>>>>>>>> Andrey
>>>>>>>>>>>
>>>>>>>>>>> [1]
>>>>>>>>>>> https://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html
>>>>>>>>>>>
>>>>>>>>>>> On 24 Aug 2018, at 20:41, Juho Autio <juho.au...@rovio.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>> Hi,
>>>>>>>>>>>
>>>>>>>>>>> Using StreamingFileSink is not a convenient option for
>>>>>>>>>>> production use for us as it doesn't support s3*. I could use
>>>>>>>>>>> StreamingFileSink just to verify, but I don't see much point in
>>>>>>>>>>> doing so.
>>>>>>>>>>> Please consider my previous comment:
>>>>>>>>>>>
>>>>>>>>>>> > I realized that BucketingSink must not play any role in this
>>>>>>>>>>> problem. This is because only when the 24-hour window triggers,
>>>>>>>>>>> BucketingSink gets a burst of input. Around the state restoring
>>>>>>>>>>> point
>>>>>>>>>>> (middle of the day) it doesn't get any input, so it can't lose
>>>>>>>>>>> anything
>>>>>>>>>>> either (right?).
>>>>>>>>>>>
>>>>>>>>>>> I could also use a kafka sink instead, but I can't imagine how
>>>>>>>>>>> there could be any difference. It's very real that the sink doesn't
>>>>>>>>>>> get any
>>>>>>>>>>> input for a long time until the 24-hour window closes, and then it
>>>>>>>>>>> quickly
>>>>>>>>>>> writes out everything because it's not that much data eventually
>>>>>>>>>>> for the
>>>>>>>>>>> distinct values.
>>>>>>>>>>>
>>>>>>>>>>> Any ideas for debugging what's happening around the savepoint &
>>>>>>>>>>> restoration time?
>>>>>>>>>>>
>>>>>>>>>>> *) I actually implemented StreamingFileSink as an alternative
>>>>>>>>>>> sink. This was before I came to realize that most likely the sink
>>>>>>>>>>> component
>>>>>>>>>>> has nothing to do with the data loss problem. I tried it with
>>>>>>>>>>> s3n:// path
>>>>>>>>>>> just to see an exception being thrown. In the source code I indeed
>>>>>>>>>>> then
>>>>>>>>>>> found an explicit check for the target path scheme to be "hdfs://".
>>>>>>>>>>>
>>>>>>>>>>> On Fri, Aug 24, 2018 at 7:49 PM Andrey Zagrebin <
>>>>>>>>>>> and...@data-artisans.com> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Ok, I think before further debugging the window reduced state,
>>>>>>>>>>>> could you try the new ‘StreamingFileSink’ [1] introduced in
>>>>>>>>>>>> Flink 1.6.0 instead of the previous 'BucketingSink’?
>>>>>>>>>>>>
>>>>>>>>>>>> Cheers,
>>>>>>>>>>>> Andrey
>>>>>>>>>>>>
>>>>>>>>>>>> [1]
>>>>>>>>>>>> https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/streamfile_sink.html
>>>>>>>>>>>>
>>>>>>>>>>>> On 24 Aug 2018, at 18:03, Juho Autio <juho.au...@rovio.com>
>>>>>>>>>>>> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>> Yes, sorry for my confusing comment. I just meant that it seems
>>>>>>>>>>>> like there's a bug somewhere now that the output is missing some
>>>>>>>>>>>> data.
>>>>>>>>>>>>
>>>>>>>>>>>> > I would wait and check the actual output in s3 because it is
>>>>>>>>>>>> the main result of the job
>>>>>>>>>>>>
>>>>>>>>>>>> Yes, and that's what I have already done. There seems to be
>>>>>>>>>>>> always some data loss with the production data volumes, if the job
>>>>>>>>>>>> has been
>>>>>>>>>>>> restarted on that day.
>>>>>>>>>>>>
>>>>>>>>>>>> Would you have any suggestions for how to debug this further?
>>>>>>>>>>>>
>>>>>>>>>>>> Many thanks for stepping in.
>>>>>>>>>>>>
>>>>>>>>>>>> On Fri, Aug 24, 2018 at 6:37 PM Andrey Zagrebin <
>>>>>>>>>>>> and...@data-artisans.com> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Hi Juho,
>>>>>>>>>>>>>
>>>>>>>>>>>>> So it is a per key deduplication job.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Yes, I would wait and check the actual output in s3 because it
>>>>>>>>>>>>> is the main result of the job and
>>>>>>>>>>>>>
>>>>>>>>>>>>> > The late data around the time of taking savepoint might be
>>>>>>>>>>>>> not included into the savepoint but it should be behind the
>>>>>>>>>>>>> snapshotted
>>>>>>>>>>>>> offset in Kafka.
>>>>>>>>>>>>>
>>>>>>>>>>>>> is not a bug, it is a possible behaviour.
>>>>>>>>>>>>>
>>>>>>>>>>>>> The savepoint is a snapshot of the data in transient which is
>>>>>>>>>>>>> already consumed from Kafka.
>>>>>>>>>>>>> Basically the full contents of the window result is split
>>>>>>>>>>>>> between the savepoint and what can come after the savepoint'ed
>>>>>>>>>>>>> offset in
>>>>>>>>>>>>> Kafka but before the window result is written into s3.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Allowed lateness should not affect it, I am just saying that
>>>>>>>>>>>>> the final result in s3 should include all records after it.
>>>>>>>>>>>>> This is what should be guaranteed but not the contents of the
>>>>>>>>>>>>> intermediate savepoint.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>> Andrey
>>>>>>>>>>>>>
>>>>>>>>>>>>> On 24 Aug 2018, at 16:52, Juho Autio <juho.au...@rovio.com>
>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>> Thanks for your answer!
>>>>>>>>>>>>>
>>>>>>>>>>>>> I check for the missed data from the final output on s3. So I
>>>>>>>>>>>>> wait until the next day, then run the same thing re-implemented
>>>>>>>>>>>>> in batch,
>>>>>>>>>>>>> and compare the output.
>>>>>>>>>>>>>
>>>>>>>>>>>>> > The late data around the time of taking savepoint might be
>>>>>>>>>>>>> not included into the savepoint but it should be behind the
>>>>>>>>>>>>> snapshotted
>>>>>>>>>>>>> offset in Kafka.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Yes, I would definitely expect that. It seems like there's a
>>>>>>>>>>>>> bug somewhere.
>>>>>>>>>>>>>
>>>>>>>>>>>>> > Then it should just come later after the restore and should
>>>>>>>>>>>>> be reduced within the allowed lateness into the final result
>>>>>>>>>>>>> which is saved
>>>>>>>>>>>>> into s3.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Well, as far as I know, allowed lateness doesn't play any role
>>>>>>>>>>>>> here, because I started running the job with allowedLateness=0,
>>>>>>>>>>>>> and still
>>>>>>>>>>>>> get the data loss, while my late data output doesn't receive
>>>>>>>>>>>>> anything.
>>>>>>>>>>>>>
>>>>>>>>>>>>> > Also, is this `DistinctFunction.reduce` just an example or
>>>>>>>>>>>>> the actual implementation, basically saving just one of records
>>>>>>>>>>>>> inside the
>>>>>>>>>>>>> 24h window in s3? then what is missing there?
>>>>>>>>>>>>>
>>>>>>>>>>>>> Yes, it's the actual implementation. Note that there's a keyBy
>>>>>>>>>>>>> before the DistinctFunction. So there's one record for each key
>>>>>>>>>>>>> (which is
>>>>>>>>>>>>> the combination of a couple of fields). In practice I've seen
>>>>>>>>>>>>> that we're
>>>>>>>>>>>>> missing ~2000-4000 elements on each restore, and the total output
>>>>>>>>>>>>> is
>>>>>>>>>>>>> obviously much more than that.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Here's the full code for the key selector:
>>>>>>>>>>>>>
>>>>>>>>>>>>> public class MapKeySelector implements
>>>>>>>>>>>>> KeySelector<Map<String,String>, Object> {
>>>>>>>>>>>>>
>>>>>>>>>>>>> private final String[] fields;
>>>>>>>>>>>>>
>>>>>>>>>>>>> public MapKeySelector(String... fields) {
>>>>>>>>>>>>> this.fields = fields;
>>>>>>>>>>>>> }
>>>>>>>>>>>>>
>>>>>>>>>>>>> @Override
>>>>>>>>>>>>> public Object getKey(Map<String, String> event) throws
>>>>>>>>>>>>> Exception {
>>>>>>>>>>>>> Tuple key =
>>>>>>>>>>>>> Tuple.getTupleClass(fields.length).newInstance();
>>>>>>>>>>>>> for (int i = 0; i < fields.length; i++) {
>>>>>>>>>>>>> key.setField(event.getOrDefault(fields[i], ""), i);
>>>>>>>>>>>>> }
>>>>>>>>>>>>> return key;
>>>>>>>>>>>>> }
>>>>>>>>>>>>> }
>>>>>>>>>>>>>
>>>>>>>>>>>>> And a more exact example on how it's used:
>>>>>>>>>>>>>
>>>>>>>>>>>>> .keyBy(new MapKeySelector("ID", "PLAYER_ID",
>>>>>>>>>>>>> "FIELD", "KEY_NAME", "KEY_VALUE"))
>>>>>>>>>>>>> .timeWindow(Time.days(1))
>>>>>>>>>>>>> .reduce(new DistinctFunction())
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Fri, Aug 24, 2018 at 5:26 PM Andrey Zagrebin <
>>>>>>>>>>>>> and...@data-artisans.com> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> Hi Juho,
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Where exactly does the data miss? When do you notice that?
>>>>>>>>>>>>>> Do you check it:
>>>>>>>>>>>>>> - debugging `DistinctFunction.reduce` right after resume in
>>>>>>>>>>>>>> the middle of the day
>>>>>>>>>>>>>> or
>>>>>>>>>>>>>> - some distinct records miss in the final output
>>>>>>>>>>>>>> of BucketingSink in s3 after window result is actually triggered
>>>>>>>>>>>>>> and saved
>>>>>>>>>>>>>> into s3 at the end of the day? is this the main output?
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> The late data around the time of taking savepoint might be
>>>>>>>>>>>>>> not included into the savepoint but it should be behind the
>>>>>>>>>>>>>> snapshotted
>>>>>>>>>>>>>> offset in Kafka. Then it should just come later after the
>>>>>>>>>>>>>> restore and
>>>>>>>>>>>>>> should be reduced within the allowed lateness into the final
>>>>>>>>>>>>>> result which
>>>>>>>>>>>>>> is saved into s3.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Also, is this `DistinctFunction.reduce` just an example or
>>>>>>>>>>>>>> the actual implementation, basically saving just one of records
>>>>>>>>>>>>>> inside the
>>>>>>>>>>>>>> 24h window in s3? then what is missing there?
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>>> Andrey
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On 23 Aug 2018, at 15:42, Juho Autio <juho.au...@rovio.com>
>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> I changed to allowedLateness=0, no change, still missing data
>>>>>>>>>>>>>> when restoring from savepoint.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Tue, Aug 21, 2018 at 10:43 AM Juho Autio <
>>>>>>>>>>>>>> juho.au...@rovio.com> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> I realized that BucketingSink must not play any role in this
>>>>>>>>>>>>>>> problem. This is because only when the 24-hour window triggers,
>>>>>>>>>>>>>>> BucketinSink gets a burst of input. Around the state restoring
>>>>>>>>>>>>>>> point
>>>>>>>>>>>>>>> (middle of the day) it doesn't get any input, so it can't lose
>>>>>>>>>>>>>>> anything
>>>>>>>>>>>>>>> either (right?).
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> I will next try removing the allowedLateness entirely from
>>>>>>>>>>>>>>> the equation.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> In the meanwhile, please let me know if you have any
>>>>>>>>>>>>>>> suggestions for debugging the lost data, for example what logs
>>>>>>>>>>>>>>> to enable.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> We use FlinkKafkaConsumer010 btw. Are there any known issues
>>>>>>>>>>>>>>> with that, that could contribute to lost data when restoring a
>>>>>>>>>>>>>>> savepoint?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On Fri, Aug 17, 2018 at 4:23 PM Juho Autio <
>>>>>>>>>>>>>>> juho.au...@rovio.com> wrote:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Some data is silently lost on my Flink stream job when
>>>>>>>>>>>>>>>> state is restored from a savepoint.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Do you have any debugging hints to find out where exactly
>>>>>>>>>>>>>>>> the data gets dropped?
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> My job gathers distinct values using a 24-hour window. It
>>>>>>>>>>>>>>>> doesn't have any custom state management.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> When I cancel the job with savepoint and restore from that
>>>>>>>>>>>>>>>> savepoint, some data is missed. It seems to be losing just a
>>>>>>>>>>>>>>>> small amount
>>>>>>>>>>>>>>>> of data. The event time of lost data is probably around the
>>>>>>>>>>>>>>>> time of
>>>>>>>>>>>>>>>> savepoint. In other words the rest of the time window is not
>>>>>>>>>>>>>>>> entirely
>>>>>>>>>>>>>>>> missed – collection works correctly also for (most of the)
>>>>>>>>>>>>>>>> events that come
>>>>>>>>>>>>>>>> in after restoring.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> When the job processes a full 24-hour window without
>>>>>>>>>>>>>>>> interruptions it doesn't miss anything.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Usually the problem doesn't happen in test environments
>>>>>>>>>>>>>>>> that have smaller parallelism and smaller data volumes. But in
>>>>>>>>>>>>>>>> production
>>>>>>>>>>>>>>>> volumes the job seems to be consistently missing at least
>>>>>>>>>>>>>>>> something on
>>>>>>>>>>>>>>>> every restore.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> This issue has consistently happened since the job was
>>>>>>>>>>>>>>>> initially created. It was at first run on an older version of
>>>>>>>>>>>>>>>> Flink
>>>>>>>>>>>>>>>> 1.5-SNAPSHOT and it still happens on both Flink 1.5.2 & 1.6.0.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> I'm wondering if this could be for example some
>>>>>>>>>>>>>>>> synchronization issue between the kafka consumer offsets vs.
>>>>>>>>>>>>>>>> what's been
>>>>>>>>>>>>>>>> written by BucketingSink?
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> 1. Job content, simplified
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> kafkaStream
>>>>>>>>>>>>>>>> .flatMap(new ExtractFieldsFunction())
>>>>>>>>>>>>>>>> .keyBy(new MapKeySelector(1, 2, 3, 4))
>>>>>>>>>>>>>>>> .timeWindow(Time.days(1))
>>>>>>>>>>>>>>>> .allowedLateness(allowedLateness)
>>>>>>>>>>>>>>>> .sideOutputLateData(lateDataTag)
>>>>>>>>>>>>>>>> .reduce(new DistinctFunction())
>>>>>>>>>>>>>>>> .addSink(sink)
>>>>>>>>>>>>>>>> // use a fixed number of output partitions
>>>>>>>>>>>>>>>> .setParallelism(8))
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> /**
>>>>>>>>>>>>>>>> * Usage: .keyBy("the", "distinct", "fields").reduce(new
>>>>>>>>>>>>>>>> DistinctFunction())
>>>>>>>>>>>>>>>> */
>>>>>>>>>>>>>>>> public class DistinctFunction implements
>>>>>>>>>>>>>>>> ReduceFunction<java.util.Map<String, String>> {
>>>>>>>>>>>>>>>> @Override
>>>>>>>>>>>>>>>> public Map<String, String> reduce(Map<String, String>
>>>>>>>>>>>>>>>> value1, Map<String, String> value2) {
>>>>>>>>>>>>>>>> return value1;
>>>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> 2. State configuration
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> boolean enableIncrementalCheckpointing = true;
>>>>>>>>>>>>>>>> String statePath = "s3n://bucket/savepoints";
>>>>>>>>>>>>>>>> new RocksDBStateBackend(statePath,
>>>>>>>>>>>>>>>> enableIncrementalCheckpointing);
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Checkpointing Mode Exactly Once
>>>>>>>>>>>>>>>> Interval 1m 0s
>>>>>>>>>>>>>>>> Timeout 10m 0s
>>>>>>>>>>>>>>>> Minimum Pause Between Checkpoints 1m 0s
>>>>>>>>>>>>>>>> Maximum Concurrent Checkpoints 1
>>>>>>>>>>>>>>>> Persist Checkpoints Externally Enabled (retain on
>>>>>>>>>>>>>>>> cancellation)
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> 3. BucketingSink configuration
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> We use BucketingSink, I don't think there's anything
>>>>>>>>>>>>>>>> special here, if not the fact that we're writing to S3.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> String outputPath = "s3://bucket/output";
>>>>>>>>>>>>>>>> BucketingSink<Map<String, String>> sink = new
>>>>>>>>>>>>>>>> BucketingSink<Map<String, String>>(outputPath)
>>>>>>>>>>>>>>>> .setBucketer(new ProcessdateBucketer())
>>>>>>>>>>>>>>>> .setBatchSize(batchSize)
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> .setInactiveBucketThreshold(inactiveBucketThreshold)
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> .setInactiveBucketCheckInterval(inactiveBucketCheckInterval);
>>>>>>>>>>>>>>>> sink.setWriter(new IdJsonWriter());
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> 4. Kafka & event time
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> My flink job reads the data from Kafka, using a
>>>>>>>>>>>>>>>> BoundedOutOfOrdernessTimestampExtractor on the kafka consumer
>>>>>>>>>>>>>>>> to
>>>>>>>>>>>>>>>> synchronize watermarks accross all kafka partitions. We also
>>>>>>>>>>>>>>>> write late
>>>>>>>>>>>>>>>> data to side output, but nothing is written there – if it
>>>>>>>>>>>>>>>> would, it could
>>>>>>>>>>>>>>>> explain missed data in the main output (I'm also sure that our
>>>>>>>>>>>>>>>> late data
>>>>>>>>>>>>>>>> writing works, because we previously had some actual late data
>>>>>>>>>>>>>>>> which ended
>>>>>>>>>>>>>>>> up there).
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> 5. allowedLateness
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> It may be or may not be relevant that I have also enabled
>>>>>>>>>>>>>>>> allowedLateness with 1 minute lateness on the 24-hour window:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> If that makes sense, I could try removing allowedLateness
>>>>>>>>>>>>>>>> entirely? That would be just to rule out that Flink doesn't
>>>>>>>>>>>>>>>> have a bug
>>>>>>>>>>>>>>>> that's related to restoring state in combination with the
>>>>>>>>>>>>>>>> allowedLateness
>>>>>>>>>>>>>>>> feature. After all, all of our data should be in a good enough
>>>>>>>>>>>>>>>> order to not
>>>>>>>>>>>>>>>> be late, given the max out of orderness used on kafka consumer
>>>>>>>>>>>>>>>> timestamp
>>>>>>>>>>>>>>>> extractor.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Thank you in advance!
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>>
>>>>
>>>>
>>>
>>>
>>
>
