flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From StefanRRichter <...@git.apache.org>
Subject [GitHub] flink pull request #5947: [FLINK-8978] Stateful generic stream job upgrade e...
Date Wed, 02 May 2018 16:00:42 GMT
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.


---

Mime
View raw message