You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2018/12/30 14:16:33 UTC
[flink] 06/08: [hotfix][cep] Fixed passing runtime context
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bec4a2885102687fd947a31870e5e1a1efb37ba8
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Tue Dec 18 09:35:40 2018 +0100
[hotfix][cep] Fixed passing runtime context
---
.../flink/cep/RichPatternFlatSelectFunction.java | 15 +-
.../flink/cep/RichPatternSelectFunction.java | 15 +-
.../functions/RichPatternFlatSelectFunction.java | 45 ------
.../cep/functions/RichPatternSelectFunction.java | 44 ------
.../main/java/org/apache/flink/cep/nfa/NFA.java | 34 +++-
.../org/apache/flink/cep/operator/CepOperator.java | 43 ++---
.../cep/{ => operator}/CepRuntimeContext.java | 54 ++++---
.../pattern/conditions/RichIterativeCondition.java | 8 +-
.../operator/CepProcessFunctionContextTest.java | 19 ++-
.../cep/{ => operator}/CepRuntimeContextTest.java | 174 +++++++++++----------
10 files changed, 188 insertions(+), 263 deletions(-)
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternFlatSelectFunction.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternFlatSelectFunction.java
index a2b89c3..7ac199e 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternFlatSelectFunction.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternFlatSelectFunction.java
@@ -20,9 +20,7 @@ package org.apache.flink.cep;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.api.common.functions.RichFunction;
-import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.Preconditions;
import java.util.List;
import java.util.Map;
@@ -42,16 +40,5 @@ public abstract class RichPatternFlatSelectFunction<IN, OUT>
private static final long serialVersionUID = 1L;
- @Override
- public void setRuntimeContext(RuntimeContext runtimeContext) {
- Preconditions.checkNotNull(runtimeContext);
-
- if (runtimeContext instanceof CepRuntimeContext) {
- super.setRuntimeContext(runtimeContext);
- } else {
- super.setRuntimeContext(new CepRuntimeContext(runtimeContext));
- }
- }
-
- public abstract void flatSelect(Map<String, List<IN>> pattern, Collector<OUT> out) throws Exception;
+ public abstract void flatSelect(final Map<String, List<IN>> pattern, final Collector<OUT> out) throws Exception;
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternSelectFunction.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternSelectFunction.java
index ce694a3..a907e2b 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternSelectFunction.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/RichPatternSelectFunction.java
@@ -20,8 +20,6 @@ package org.apache.flink.cep;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.api.common.functions.RichFunction;
-import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.util.Preconditions;
import java.util.List;
import java.util.Map;
@@ -41,16 +39,5 @@ public abstract class RichPatternSelectFunction<IN, OUT>
private static final long serialVersionUID = 1L;
- @Override
- public void setRuntimeContext(RuntimeContext runtimeContext) {
- Preconditions.checkNotNull(runtimeContext);
-
- if (runtimeContext instanceof CepRuntimeContext) {
- super.setRuntimeContext(runtimeContext);
- } else {
- super.setRuntimeContext(new CepRuntimeContext(runtimeContext));
- }
- }
-
- public abstract OUT select(Map<String, List<IN>> pattern) throws Exception;
+ public abstract OUT select(final Map<String, List<IN>> pattern) throws Exception;
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/RichPatternFlatSelectFunction.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/RichPatternFlatSelectFunction.java
deleted file mode 100644
index fdc5640..0000000
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/RichPatternFlatSelectFunction.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.cep.functions;
-
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.RichFunction;
-import org.apache.flink.cep.PatternFlatSelectFunction;
-import org.apache.flink.util.Collector;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * Rich variant of the {@link PatternFlatSelectFunction}. As a {@link RichFunction}, it gives access to the
- * {@link org.apache.flink.api.common.functions.RuntimeContext} and provides setup and teardown methods:
- * {@link RichFunction#open(org.apache.flink.configuration.Configuration)} and
- * {@link RichFunction#close()}.
- *
- * @param <IN> Type of the input elements
- * @param <OUT> Type of the output element
- */
-public abstract class RichPatternFlatSelectFunction<IN, OUT>
- extends AbstractRichFunction
- implements PatternFlatSelectFunction<IN, OUT> {
-
- private static final long serialVersionUID = 1L;
-
- public abstract void flatSelect(final Map<String, List<IN>> pattern, final Collector<OUT> out) throws Exception;
-}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/RichPatternSelectFunction.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/RichPatternSelectFunction.java
deleted file mode 100644
index a4bf327..0000000
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/RichPatternSelectFunction.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.cep.functions;
-
-import org.apache.flink.api.common.functions.AbstractRichFunction;
-import org.apache.flink.api.common.functions.RichFunction;
-import org.apache.flink.cep.PatternSelectFunction;
-
-import java.util.List;
-import java.util.Map;
-
-/**
- * Rich variant of the {@link PatternSelectFunction}. As a {@link RichFunction}, it gives access to the
- * {@link org.apache.flink.api.common.functions.RuntimeContext} and provides setup and teardown methods:
- * {@link RichFunction#open(org.apache.flink.configuration.Configuration)} and
- * {@link RichFunction#close()}.
- *
- * @param <IN> Type of the input elements
- * @param <OUT> Type of the output element
- */
-public abstract class RichPatternSelectFunction<IN, OUT>
- extends AbstractRichFunction
- implements PatternSelectFunction<IN, OUT> {
-
- private static final long serialVersionUID = 1L;
-
- public abstract OUT select(final Map<String, List<IN>> pattern) throws Exception;
-}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
index b1ddb85..48bb587 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
@@ -19,6 +19,8 @@
package org.apache.flink.cep.nfa;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.typeutils.CompatibilityResult;
import org.apache.flink.api.common.typeutils.CompatibilityUtil;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerConfigSnapshot;
@@ -36,6 +38,7 @@ import org.apache.flink.cep.nfa.sharedbuffer.NodeId;
import org.apache.flink.cep.nfa.sharedbuffer.SharedBuffer;
import org.apache.flink.cep.nfa.sharedbuffer.SharedBufferAccessor;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.streaming.api.windowing.time.Time;
@@ -169,6 +172,34 @@ public class NFA<T> {
}
/**
+ * Initialization method for the NFA. It is called before any element is passed and thus suitable for one time setup
+ * work.
+ * @param cepRuntimeContext runtime context of the enclosing operator
+ * @param conf The configuration containing the parameters attached to the contract.
+ */
+ public void open(RuntimeContext cepRuntimeContext, Configuration conf) throws Exception {
+ for (State<T> state : getStates()) {
+ for (StateTransition<T> transition : state.getStateTransitions()) {
+ IterativeCondition condition = transition.getCondition();
+ FunctionUtils.setFunctionRuntimeContext(condition, cepRuntimeContext);
+ FunctionUtils.openFunction(condition, conf);
+ }
+ }
+ }
+
+ /**
+ * Tear-down method for the NFA.
+ */
+ public void close() throws Exception {
+ for (State<T> state : getStates()) {
+ for (StateTransition<T> transition : state.getStateTransitions()) {
+ IterativeCondition condition = transition.getCondition();
+ FunctionUtils.closeFunction(condition);
+ }
+ }
+ }
+
+ /**
* Processes the next input event. If some of the computations reach a final state then the
* resulting event sequences are returned. If computations time out and timeout handling is
* activated, then the timed out event patterns are returned.
@@ -224,7 +255,7 @@ public class NFA<T> {
/**
* Prunes states assuming there will be no events with timestamp <b>lower</b> than the given one.
- * It cleares the sharedBuffer and also emits all timed out partial matches.
+ * It clears the sharedBuffer and also emits all timed out partial matches.
*
* @param sharedBufferAccessor the accessor to SharedBuffer object that we need to work upon while processing
* @param nfaState The NFAState object that we need to affect while processing
@@ -264,7 +295,6 @@ public class NFA<T> {
sharedBufferAccessor.advanceTime(timestamp);
return timeoutResult;
-
}
private boolean isStateTimedOut(final ComputationState state, final long timestamp) {
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
index 0fd7053..6df1fa0 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java
@@ -36,25 +36,25 @@ import org.apache.flink.cep.nfa.NFA;
import org.apache.flink.cep.nfa.NFA.MigratedNFA;
import org.apache.flink.cep.nfa.NFAState;
import org.apache.flink.cep.nfa.NFAStateSerializer;
-import org.apache.flink.cep.nfa.State;
-import org.apache.flink.cep.nfa.StateTransition;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.nfa.compiler.NFACompiler;
import org.apache.flink.cep.nfa.sharedbuffer.SharedBuffer;
import org.apache.flink.cep.nfa.sharedbuffer.SharedBufferAccessor;
-import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.KeyedStateFunction;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
+import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.InternalTimer;
import org.apache.flink.streaming.api.operators.InternalTimerService;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.Output;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.api.operators.Triggerable;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.Preconditions;
@@ -127,6 +127,9 @@ public class CepOperator<IN, KEY, OUT>
/** Main output collector, that sets a proper timestamp to the StreamRecord. */
private transient TimestampedCollector<OUT> collector;
+ /** Wrapped RuntimeContext that limits the underlying context features. */
+ private transient CepRuntimeContext cepRuntimeContext;
+
public CepOperator(
final TypeSerializer<IN> inputSerializer,
final boolean isProcessingTime,
@@ -152,6 +155,13 @@ public class CepOperator<IN, KEY, OUT>
}
@Override
+ public void setup(StreamTask<?, ?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output) {
+ super.setup(containingTask, config, output);
+ this.cepRuntimeContext = new CepRuntimeContext(getRuntimeContext());
+ FunctionUtils.setFunctionRuntimeContext(getUserFunction(), this.cepRuntimeContext);
+ }
+
+ @Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
@@ -201,8 +211,8 @@ public class CepOperator<IN, KEY, OUT>
VoidNamespaceSerializer.INSTANCE,
this);
- this.nfa = nfaFactory.createNFA();
- openNFA(nfa);
+ nfa = nfaFactory.createNFA();
+ nfa.open(cepRuntimeContext, new Configuration());
context = new ContextFunctionImpl();
collector = new TimestampedCollector<>(output);
@@ -211,8 +221,7 @@ public class CepOperator<IN, KEY, OUT>
@Override
public void close() throws Exception {
super.close();
-
- closeNFA(nfa);
+ nfa.close();
}
@Override
@@ -498,26 +507,6 @@ public class CepOperator<IN, KEY, OUT>
}
}
- private void openNFA(NFA<IN> nfa) throws Exception {
- Configuration conf = new Configuration();
- for (State<IN> state : nfa.getStates()) {
- for (StateTransition<IN> transition : state.getStateTransitions()) {
- IterativeCondition condition = transition.getCondition();
- FunctionUtils.setFunctionRuntimeContext(condition, getRuntimeContext());
- FunctionUtils.openFunction(condition, conf);
- }
- }
- }
-
- private void closeNFA(NFA<IN> nfa) throws Exception {
- for (State<IN> state : nfa.getStates()) {
- for (StateTransition<IN> transition : state.getStateTransitions()) {
- IterativeCondition condition = transition.getCondition();
- FunctionUtils.closeFunction(condition);
- }
- }
- }
-
////////////////////// Testing Methods //////////////////////
@VisibleForTesting
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java
similarity index 75%
rename from flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java
rename to flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java
index ed93f8e..0518dad 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CepRuntimeContext.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepRuntimeContext.java
@@ -16,8 +16,9 @@
* limitations under the License.
*/
-package org.apache.flink.cep;
+package org.apache.flink.cep.operator;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.DoubleCounter;
@@ -45,17 +46,23 @@ import java.io.Serializable;
import java.util.List;
import java.util.Map;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
- * A wrapper class for pattern select function and iterative condition function's {@link RuntimeContext}.
- * The runtime context only supports basic operations. Consequently, state access, accumulators,
- * broadcast variables and the distributed cache are disabled.
+ * A wrapper class for the {@link RuntimeContext}.
+ *
+ * <p>This context only exposes the functionality needed by the
+ * pattern process function and iterative condition function.
+ * Consequently, state access, accumulators, broadcast variables
+ * and the distributed cache are disabled.
*/
-public class CepRuntimeContext implements RuntimeContext {
+@Internal
+class CepRuntimeContext implements RuntimeContext {
private final RuntimeContext runtimeContext;
- public CepRuntimeContext(RuntimeContext runtimeContext) {
- this.runtimeContext = runtimeContext;
+ CepRuntimeContext(final RuntimeContext runtimeContext) {
+ this.runtimeContext = checkNotNull(runtimeContext);
}
@Override
@@ -114,12 +121,13 @@ public class CepRuntimeContext implements RuntimeContext {
@Override
public <V, A extends Serializable> void addAccumulator(
- String name, Accumulator<V, A> accumulator) {
+ final String name,
+ final Accumulator<V, A> accumulator) {
throw new UnsupportedOperationException("Accumulators are not supported.");
}
@Override
- public <V, A extends Serializable> Accumulator<V, A> getAccumulator(String name) {
+ public <V, A extends Serializable> Accumulator<V, A> getAccumulator(final String name) {
throw new UnsupportedOperationException("Accumulators are not supported.");
}
@@ -129,68 +137,70 @@ public class CepRuntimeContext implements RuntimeContext {
}
@Override
- public IntCounter getIntCounter(String name) {
+ public IntCounter getIntCounter(final String name) {
throw new UnsupportedOperationException("Int counters are not supported.");
}
@Override
- public LongCounter getLongCounter(String name) {
+ public LongCounter getLongCounter(final String name) {
throw new UnsupportedOperationException("Long counters are not supported.");
}
@Override
- public DoubleCounter getDoubleCounter(String name) {
+ public DoubleCounter getDoubleCounter(final String name) {
throw new UnsupportedOperationException("Double counters are not supported.");
}
@Override
- public Histogram getHistogram(String name) {
+ public Histogram getHistogram(final String name) {
throw new UnsupportedOperationException("Histograms are not supported.");
}
@Override
- public boolean hasBroadcastVariable(String name) {
+ public boolean hasBroadcastVariable(final String name) {
throw new UnsupportedOperationException("Broadcast variables are not supported.");
}
@Override
- public <RT> List<RT> getBroadcastVariable(String name) {
+ public <RT> List<RT> getBroadcastVariable(final String name) {
throw new UnsupportedOperationException("Broadcast variables are not supported.");
}
@Override
public <T, C> C getBroadcastVariableWithInitializer(
- String name, BroadcastVariableInitializer<T, C> initializer) {
+ final String name,
+ final BroadcastVariableInitializer<T, C> initializer) {
throw new UnsupportedOperationException("Broadcast variables are not supported.");
}
@Override
- public <T> ValueState<T> getState(ValueStateDescriptor<T> stateProperties) {
+ public <T> ValueState<T> getState(final ValueStateDescriptor<T> stateProperties) {
throw new UnsupportedOperationException("State is not supported.");
}
@Override
- public <T> ListState<T> getListState(ListStateDescriptor<T> stateProperties) {
+ public <T> ListState<T> getListState(final ListStateDescriptor<T> stateProperties) {
throw new UnsupportedOperationException("State is not supported.");
}
@Override
- public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {
+ public <T> ReducingState<T> getReducingState(final ReducingStateDescriptor<T> stateProperties) {
throw new UnsupportedOperationException("State is not supported.");
}
@Override
- public <IN, ACC, OUT> AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT> stateProperties) {
+ public <IN, ACC, OUT> AggregatingState<IN, OUT> getAggregatingState(
+ final AggregatingStateDescriptor<IN, ACC, OUT> stateProperties) {
throw new UnsupportedOperationException("State is not supported.");
}
@Override
- public <T, ACC> FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC> stateProperties) {
+ public <T, ACC> FoldingState<T, ACC> getFoldingState(final FoldingStateDescriptor<T, ACC> stateProperties) {
throw new UnsupportedOperationException("State is not supported.");
}
@Override
- public <UK, UV> MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties) {
+ public <UK, UV> MapState<UK, UV> getMapState(final MapStateDescriptor<UK, UV> stateProperties) {
throw new UnsupportedOperationException("State is not supported.");
}
}
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichIterativeCondition.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichIterativeCondition.java
index 12f017e..da20321 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichIterativeCondition.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/RichIterativeCondition.java
@@ -21,7 +21,6 @@ package org.apache.flink.cep.pattern.conditions;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.cep.CepRuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Preconditions;
@@ -46,12 +45,7 @@ public abstract class RichIterativeCondition<T>
@Override
public void setRuntimeContext(RuntimeContext runtimeContext) {
Preconditions.checkNotNull(runtimeContext);
-
- if (runtimeContext instanceof CepRuntimeContext) {
- this.runtimeContext = runtimeContext;
- } else {
- this.runtimeContext = new CepRuntimeContext(runtimeContext);
- }
+ this.runtimeContext = runtimeContext;
}
@Override
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java
index f35d08e..0947748 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepProcessFunctionContextTest.java
@@ -45,6 +45,9 @@ import static org.apache.flink.cep.utils.OutputAsserter.assertOutput;
*/
public class CepProcessFunctionContextTest extends TestLogger {
+ private static final boolean PROCESSING_TIME = false;
+ private static final boolean EVENT_TIME = true;
+
@Test
public void testTimestampPassingInEventTime() throws Exception {
@@ -53,7 +56,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractTimestampAndNames(1),
new NFAForwardingFactory(),
- false))) {
+ PROCESSING_TIME))) {
harness.open();
// events out of order to test if internal sorting does not mess up the timestamps
@@ -78,7 +81,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractTimestampAndNames(1),
new NFAForwardingFactory(),
- true))) {
+ EVENT_TIME))) {
harness.open();
harness.processElement(event().withName("A").withTimestamp(5).asStreamRecord());
@@ -99,7 +102,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractCurrentProcessingTimeAndNames(1),
new NFAForwardingFactory(),
- true))) {
+ EVENT_TIME))) {
harness.open();
harness.setProcessingTime(15);
@@ -122,7 +125,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractCurrentProcessingTimeAndNames(1),
new NFAForwardingFactory(),
- false))) {
+ PROCESSING_TIME))) {
harness.open();
harness.setProcessingTime(10);
@@ -147,7 +150,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractTimestampAndNames(2, timedOut),
new NFATimingOutFactory(),
- false))) {
+ PROCESSING_TIME))) {
harness.open();
// events out of order to test if internal sorting does not mess up the timestamps
@@ -178,7 +181,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractTimestampAndNames(2, timedOut),
new NFATimingOutFactory(),
- true))) {
+ EVENT_TIME))) {
harness.open();
harness.setProcessingTime(3);
@@ -208,7 +211,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractCurrentProcessingTimeAndNames(2, sideOutputTag),
new NFATimingOutFactory(),
- false))) {
+ PROCESSING_TIME))) {
harness.open();
// events out of order to test if internal sorting does not mess up the timestamps
@@ -240,7 +243,7 @@ public class CepProcessFunctionContextTest extends TestLogger {
createCepOperator(
extractCurrentProcessingTimeAndNames(2, sideOutputTag),
new NFATimingOutFactory(),
- true))) {
+ EVENT_TIME))) {
harness.open();
harness.setProcessingTime(3);
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CepRuntimeContextTest.java b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java
similarity index 66%
rename from flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CepRuntimeContextTest.java
rename to flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java
index ef7ee898..b77dd94 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/CepRuntimeContextTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CepRuntimeContextTest.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.flink.cep;
+package org.apache.flink.cep.operator;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -25,7 +25,6 @@ import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.state.FoldingStateDescriptor;
@@ -33,100 +32,84 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueStateDescriptor;
-import org.apache.flink.cep.pattern.conditions.RichAndCondition;
-import org.apache.flink.cep.pattern.conditions.RichCompositeIterativeCondition;
-import org.apache.flink.cep.pattern.conditions.RichIterativeCondition;
-import org.apache.flink.cep.pattern.conditions.RichNotCondition;
-import org.apache.flink.cep.pattern.conditions.RichOrCondition;
+import org.apache.flink.cep.Event;
+import org.apache.flink.cep.functions.PatternProcessFunction;
+import org.apache.flink.cep.nfa.NFA;
+import org.apache.flink.cep.nfa.compiler.NFACompiler;
+import org.apache.flink.cep.pattern.Pattern;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.TestLogger;
import org.junit.Test;
import java.util.List;
import java.util.Map;
+import static org.apache.flink.cep.operator.CepOperatorTestUtilities.getCepTestHarness;
+import static org.apache.flink.cep.operator.CepRuntimeContextTest.MockProcessFunctionAsserter.assertFunction;
+import static org.apache.flink.cep.utils.CepOperatorBuilder.createOperatorForNFA;
+import static org.apache.flink.cep.utils.EventBuilder.event;
+import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* Test cases for {@link CepRuntimeContext}.
*/
-public class CepRuntimeContextTest {
+public class CepRuntimeContextTest extends TestLogger {
@Test
- public void testRichCompositeIterativeCondition() throws Exception {
- RichIterativeCondition<Integer> first = new TestRichIterativeCondition();
- RichIterativeCondition<Integer> second = new TestRichIterativeCondition();
- RichIterativeCondition<Integer> third = new TestRichIterativeCondition();
-
- RichCompositeIterativeCondition function = new RichCompositeIterativeCondition(first, second, third) {
- @Override
- public boolean filter(Object value, Context ctx) throws Exception {
- return false;
- }
- };
- function.setRuntimeContext(mock(RuntimeContext.class));
-
- assertTrue(first.getRuntimeContext() instanceof CepRuntimeContext);
- assertTrue(second.getRuntimeContext() instanceof CepRuntimeContext);
- assertTrue(third.getRuntimeContext() instanceof CepRuntimeContext);
- }
+ public void testCepRuntimeContextIsSetInNFA() throws Exception {
- @Test
- public void testRichAndCondition() throws Exception {
- RichIterativeCondition<Integer> left = new TestRichIterativeCondition();
- RichIterativeCondition<Integer> right = new TestRichIterativeCondition();
+ @SuppressWarnings("unchecked")
+ final NFA<Event> mockNFA = mock(NFA.class);
- RichAndCondition function = new RichAndCondition<>(left, right);
- function.setRuntimeContext(mock(RuntimeContext.class));
+ try (
+ OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness = getCepTestHarness(
+ createOperatorForNFA(mockNFA).build())) {
- assertTrue(left.getRuntimeContext() instanceof CepRuntimeContext);
- assertTrue(right.getRuntimeContext() instanceof CepRuntimeContext);
+ harness.open();
+ verify(mockNFA).open(any(CepRuntimeContext.class), any(Configuration.class));
+ }
}
@Test
- public void testRichOrCondition() throws Exception {
- RichIterativeCondition<Integer> left = new TestRichIterativeCondition();
- RichIterativeCondition<Integer> right = new TestRichIterativeCondition();
-
- RichOrCondition function = new RichOrCondition<>(left, right);
- function.setRuntimeContext(mock(RuntimeContext.class));
-
- assertTrue(left.getRuntimeContext() instanceof CepRuntimeContext);
- assertTrue(right.getRuntimeContext() instanceof CepRuntimeContext);
- }
+ public void testCepRuntimeContextIsSetInProcessFunction() throws Exception {
- @Test
- public void testRichNotCondition() {
- RichIterativeCondition<Integer> original = new TestRichIterativeCondition();
+ final VerifyRuntimeContextProcessFunction processFunction = new VerifyRuntimeContextProcessFunction();
- RichNotCondition function = new RichNotCondition<>(original);
- function.setRuntimeContext(mock(RuntimeContext.class));
+ try (
+ OneInputStreamOperatorTestHarness<Event, Event> harness = getCepTestHarness(
+ createOperatorForNFA(getSingleElementAlwaysTrueNFA())
+ .withFunction(processFunction)
+ .build())) {
- assertTrue(original.getRuntimeContext() instanceof CepRuntimeContext);
- }
+ harness.open();
+ Event record = event().withName("A").build();
+ harness.processElement(record, 0);
- @Test
- public void testRichPatternSelectFunction() {
- verifyRuntimeContext(new TestRichPatternSelectFunction());
+ assertFunction(processFunction)
+ .checkOpenCalled()
+ .checkCloseCalled()
+ .checkProcessMatchCalled();
+ }
}
- @Test
- public void testRichPatternFlatSelectFunction() {
- verifyRuntimeContext(new TestRichPatternFlatSelectFunction());
+ private NFA<Event> getSingleElementAlwaysTrueNFA() {
+ return NFACompiler.compileFactory(Pattern.<Event>begin("A"), false).createNFA();
}
@Test
- public void testRichIterativeCondition() {
- verifyRuntimeContext(new TestRichIterativeCondition());
- }
-
- private void verifyRuntimeContext(final RichFunction function) {
+ public void testCepRuntimeContext() {
final String taskName = "foobarTask";
final MetricGroup metricGroup = new UnregisteredMetricsGroup();
final int numberOfParallelSubtasks = 42;
@@ -149,11 +132,8 @@ public class CepRuntimeContextTest {
when(mockedRuntimeContext.getUserCodeClassLoader()).thenReturn(userCodeClassLoader);
when(mockedRuntimeContext.getDistributedCache()).thenReturn(distributedCache);
- function.setRuntimeContext(mockedRuntimeContext);
-
- RuntimeContext runtimeContext = function.getRuntimeContext();
+ RuntimeContext runtimeContext = new CepRuntimeContext(mockedRuntimeContext);
- assertTrue(runtimeContext instanceof CepRuntimeContext);
assertEquals(taskName, runtimeContext.getTaskName());
assertEquals(metricGroup, runtimeContext.getMetricGroup());
assertEquals(numberOfParallelSubtasks, runtimeContext.getNumberOfParallelSubtasks());
@@ -289,30 +269,64 @@ public class CepRuntimeContextTest {
}
}
- private static class TestRichIterativeCondition extends RichIterativeCondition<Integer> {
- private static final long serialVersionUID = 1L;
+ /* Test Utils */
+ static class MockProcessFunctionAsserter {
+ private final VerifyRuntimeContextProcessFunction function;
- @Override
- public boolean filter(Integer value, Context<Integer> ctx) throws Exception {
- return false;
+ static MockProcessFunctionAsserter assertFunction(VerifyRuntimeContextProcessFunction function) {
+ return new MockProcessFunctionAsserter(function);
+ }
+
+ private MockProcessFunctionAsserter(VerifyRuntimeContextProcessFunction function) {
+ this.function = function;
+ }
+
+ MockProcessFunctionAsserter checkOpenCalled() {
+ assertThat(function.openCalled, is(true));
+ return this;
+ }
+
+ MockProcessFunctionAsserter checkCloseCalled() {
+ assertThat(function.openCalled, is(true));
+ return this;
+ }
+
+ MockProcessFunctionAsserter checkProcessMatchCalled() {
+ assertThat(function.processMatchCalled, is(true));
+ return this;
}
}
- private static class TestRichPatternSelectFunction extends RichPatternSelectFunction<Integer, Integer> {
- private static final long serialVersionUID = 1L;
+ private static class VerifyRuntimeContextProcessFunction extends PatternProcessFunction<Event, Event> {
+
+ boolean openCalled = false;
+ boolean closeCalled = false;
+ boolean processMatchCalled = false;
@Override
- public Integer select(Map<String, List<Integer>> pattern) throws Exception {
- return null;
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ verifyContext();
+ openCalled = true;
+ }
+
+ private void verifyContext() {
+ if (!(getRuntimeContext() instanceof CepRuntimeContext)) {
+ fail("Runtime context was not wrapped in CepRuntimeContext");
+ }
}
- }
- private static class TestRichPatternFlatSelectFunction extends RichPatternFlatSelectFunction<Integer, Integer> {
- private static final long serialVersionUID = 1L;
+ @Override
+ public void close() throws Exception {
+ super.close();
+ verifyContext();
+ closeCalled = true;
+ }
@Override
- public void flatSelect(Map<String, List<Integer>> pattern, Collector<Integer> out) throws Exception {
- // no op
+ public void processMatch(Map<String, List<Event>> match, Context ctx, Collector<Event> out) throws Exception {
+ verifyContext();
+ processMatchCalled = true;
}
}
}