flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-8978) End-to-end test: Job upgrade
Date Wed, 02 May 2018 16:01:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-8978?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16461225#comment-16461225
] 

ASF GitHub Bot commented on FLINK-8978:
---------------------------------------

Github user StefanRRichter commented on a diff in the pull request:

    https://github.com/apache/flink/pull/5947#discussion_r185541401
  
    --- Diff: flink-end-to-end-tests/flink-stream-stateful-job-upgrade-test/src/main/java/org/apache/flink/streaming/tests/StatefulStreamJobUpgradeTestProgram.java
---
    @@ -0,0 +1,128 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you under the Apache License, Version 2.0 (the
    + * "License"); you may not use this file except in compliance
    + * with the License.  You may obtain a copy of the License at
    + *
    + * http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.flink.streaming.tests;
    +
    +import org.apache.flink.api.common.functions.JoinFunction;
    +import org.apache.flink.api.common.functions.MapFunction;
    +import org.apache.flink.api.common.typeutils.TypeSerializer;
    +import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
    +import org.apache.flink.api.java.utils.ParameterTool;
    +import org.apache.flink.configuration.ConfigOption;
    +import org.apache.flink.configuration.ConfigOptions;
    +import org.apache.flink.streaming.api.datastream.KeyedStream;
    +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    +import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
    +import org.apache.flink.streaming.tests.artificialstate.eventpayload.ComplexPayload;
    +
    +import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createArtificialKeyedStateMapper;
    +import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createEventSource;
    +import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createSemanticsCheckMapper;
    +import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.createTimestampExtractor;
    +import static org.apache.flink.streaming.tests.DataStreamAllroundTestJobFactory.setupEnvironment;
    +
    +import java.util.Collections;
    +import java.util.List;
    +
    +/**
    + * Test upgrade of generic stateful job for Flink's DataStream API operators and primitives.
    + *
    + * <p>The job is constructed of generic components from {@link DataStreamAllroundTestJobFactory}.
    + * The gaol is to test successful state restoration after taking savepoint and recovery
with new job version.
    + * It can be configured with '--test.job.variant' to run different variants of it:
    + * <ul>
    + *     <li><b>original:</b> includes 2 custom stateful map operators</li>
    + *     <li><b>upgraded:</b> changes order of 2 custom stateful map
operators and adds one more</li>
    + * </ul>
    + */
    +public class StatefulStreamJobUpgradeTestProgram {
    +	private static final String TEST_JOB_VARIANT_ORIGINAL = "original";
    +	private static final String TEST_JOB_VARIANT_UPGRADED = "upgraded";
    +
    +	private static final JoinFunction<Event, ComplexPayload, ComplexPayload> SIMPLE_STATE_UPDATE
=
    +		(Event first, ComplexPayload second) -> new ComplexPayload(first);
    +	private static final JoinFunction<Event, ComplexPayload, ComplexPayload> LAST_EVENT_STATE_UPDATE
=
    +		(Event first, ComplexPayload second) ->
    +			(second != null && first.getEventTime() <= second.getEventTime()) ? second
: new ComplexPayload(first);
    +
    +	private static final ConfigOption<String> TEST_JOB_VARIANT = ConfigOptions
    +		.key("test.job.variant")
    +		.defaultValue(TEST_JOB_VARIANT_ORIGINAL)
    +		.withDescription(String.format("This configures the job variant to test. Can be '%s'
or '%s'",
    +			TEST_JOB_VARIANT_ORIGINAL, TEST_JOB_VARIANT_UPGRADED));
    +
    +	public static void main(String[] args) throws Exception {
    +		final ParameterTool pt = ParameterTool.fromArgs(args);
    +
    +		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    +
    +		setupEnvironment(env, pt);
    +
    +		KeyedStream<Event, Integer> source = env.addSource(createEventSource(pt))
    +			.assignTimestampsAndWatermarks(createTimestampExtractor(pt))
    +			.keyBy(Event::getKey);
    +
    +		List<TypeSerializer<ComplexPayload>> stateSer =
    +			Collections.singletonList(new KryoSerializer<>(ComplexPayload.class, env.getConfig()));
    +
    +		boolean isOriginal = pt.get(TEST_JOB_VARIANT.key()).equals(TEST_JOB_VARIANT_ORIGINAL);
    --- End diff --
    
    We could throw an `IllegalArgumentException` for any unexpected string.


> End-to-end test: Job upgrade
> ----------------------------
>
>                 Key: FLINK-8978
>                 URL: https://issues.apache.org/jira/browse/FLINK-8978
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Tests
>    Affects Versions: 1.5.0
>            Reporter: Till Rohrmann
>            Assignee: Andrey Zagrebin
>            Priority: Blocker
>             Fix For: 1.6.0, 1.5.1
>
>
> Job upgrades usually happen during the lifetime of a real world Flink job. Therefore,
we should add an end-to-end test which exactly covers this scenario. I suggest to do the follwoing:
> # run the general purpose testing job FLINK-8971
> # take a savepoint
> # Modify the job by introducing a new operator and changing the order of others
> # Resume the modified job from the savepoint
> # Verify that everything went correctly



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Mime
View raw message