flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kklou...@apache.org
Subject [5/9] flink git commit: [hotfix] [cep] Remove unused keySelector in operator.
Date Wed, 17 May 2017 12:42:45 GMT
[hotfix] [cep] Remove unused keySelector in operator.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/34a6020f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/34a6020f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/34a6020f

Branch: refs/heads/release-1.3
Commit: 34a6020ff6d51cd8148366677fadb9317cfaa153
Parents: 36df901
Author: kkloudas <kkloudas@gmail.com>
Authored: Mon May 15 14:49:00 2017 +0200
Committer: kkloudas <kkloudas@gmail.com>
Committed: Wed May 17 14:40:26 2017 +0200

----------------------------------------------------------------------
 .../cep/operator/AbstractKeyedCEPPatternOperator.java   |  6 ------
 .../org/apache/flink/cep/operator/CEPOperatorUtils.java |  6 ------
 .../flink/cep/operator/KeyedCEPPatternOperator.java     |  4 +---
 .../cep/operator/TimeoutKeyedCEPPatternOperator.java    |  4 +---
 .../flink/cep/operator/CEPFrom12MigrationTest.java      |  6 ------
 .../flink/cep/operator/CEPMigration11to13Test.java      |  2 --
 .../org/apache/flink/cep/operator/CEPOperatorTest.java  | 12 ++++--------
 .../org/apache/flink/cep/operator/CEPRescalingTest.java |  1 -
 8 files changed, 6 insertions(+), 35 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
index 7068bc4..bac21b3 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/AbstractKeyedCEPPatternOperator.java
@@ -26,7 +26,6 @@ import org.apache.flink.api.common.typeutils.TypeDeserializerAdapter;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
 import org.apache.flink.api.common.typeutils.base.CollectionSerializerConfigSnapshot;
-import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.cep.nfa.NFA;
 import org.apache.flink.cep.nfa.compiler.NFACompiler;
 import org.apache.flink.core.fs.FSDataInputStream;
@@ -78,9 +77,6 @@ public abstract class AbstractKeyedCEPPatternOperator<IN, KEY, OUT>
 
 	private final TypeSerializer<IN> inputSerializer;
 
-	// necessary to extract the key from the input elements
-	private final KeySelector<IN, KEY> keySelector;
-
 	// necessary to serialize the set of seen keys
 	private final TypeSerializer<KEY> keySerializer;
 
@@ -112,14 +108,12 @@ public abstract class AbstractKeyedCEPPatternOperator<IN, KEY, OUT>
 	public AbstractKeyedCEPPatternOperator(
 			final TypeSerializer<IN> inputSerializer,
 			final boolean isProcessingTime,
-			final KeySelector<IN, KEY> keySelector,
 			final TypeSerializer<KEY> keySerializer,
 			final NFACompiler.NFAFactory<IN> nfaFactory,
 			final boolean migratingFromOldKeyedOperator) {
 
 		this.inputSerializer = Preconditions.checkNotNull(inputSerializer);
 		this.isProcessingTime = Preconditions.checkNotNull(isProcessingTime);
-		this.keySelector = Preconditions.checkNotNull(keySelector);
 		this.keySerializer = Preconditions.checkNotNull(keySerializer);
 		this.nfaFactory = Preconditions.checkNotNull(nfaFactory);
 

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
index 08424a4..e7b7e65 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPOperatorUtils.java
@@ -63,7 +63,6 @@ public class CEPOperatorUtils {
 			// We have to use the KeyedCEPPatternOperator which can deal with keyed input streams
 			KeyedStream<T, K> keyedStream= (KeyedStream<T, K>) inputStream;
 
-			KeySelector<T, K> keySelector = keyedStream.getKeySelector();
 			TypeSerializer<K> keySerializer = keyedStream.getKeyType().createSerializer(keyedStream.getExecutionConfig());
 
 			patternStream = keyedStream.transform(
@@ -72,7 +71,6 @@ public class CEPOperatorUtils {
 				new KeyedCEPPatternOperator<>(
 					inputSerializer,
 					isProcessingTime,
-					keySelector,
 					keySerializer,
 					nfaFactory,
 					true));
@@ -87,7 +85,6 @@ public class CEPOperatorUtils {
 				new KeyedCEPPatternOperator<>(
 					inputSerializer,
 					isProcessingTime,
-					keySelector,
 					keySerializer,
 					nfaFactory,
 					false
@@ -127,7 +124,6 @@ public class CEPOperatorUtils {
 			// We have to use the KeyedCEPPatternOperator which can deal with keyed input streams
 			KeyedStream<T, K> keyedStream= (KeyedStream<T, K>) inputStream;
 
-			KeySelector<T, K> keySelector = keyedStream.getKeySelector();
 			TypeSerializer<K> keySerializer = keyedStream.getKeyType().createSerializer(keyedStream.getExecutionConfig());
 
 			patternStream = keyedStream.transform(
@@ -136,7 +132,6 @@ public class CEPOperatorUtils {
 				new TimeoutKeyedCEPPatternOperator<>(
 					inputSerializer,
 					isProcessingTime,
-					keySelector,
 					keySerializer,
 					nfaFactory,
 					true));
@@ -151,7 +146,6 @@ public class CEPOperatorUtils {
 				new TimeoutKeyedCEPPatternOperator<>(
 					inputSerializer,
 					isProcessingTime,
-					keySelector,
 					keySerializer,
 					nfaFactory,
 					false

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/KeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/KeyedCEPPatternOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/KeyedCEPPatternOperator.java
index 4d68afb..fec226a 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/KeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/KeyedCEPPatternOperator.java
@@ -19,7 +19,6 @@
 package org.apache.flink.cep.operator;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.cep.nfa.NFA;
 import org.apache.flink.cep.nfa.compiler.NFACompiler;
@@ -44,12 +43,11 @@ public class KeyedCEPPatternOperator<IN, KEY> extends AbstractKeyedCEPPatternOpe
 	public KeyedCEPPatternOperator(
 			TypeSerializer<IN> inputSerializer,
 			boolean isProcessingTime,
-			KeySelector<IN, KEY> keySelector,
 			TypeSerializer<KEY> keySerializer,
 			NFACompiler.NFAFactory<IN> nfaFactory,
 			boolean migratingFromOldKeyedOperator) {
 
-		super(inputSerializer, isProcessingTime, keySelector, keySerializer, nfaFactory, migratingFromOldKeyedOperator);
+		super(inputSerializer, isProcessingTime, keySerializer, nfaFactory, migratingFromOldKeyedOperator);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/TimeoutKeyedCEPPatternOperator.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/TimeoutKeyedCEPPatternOperator.java
b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/TimeoutKeyedCEPPatternOperator.java
index 9061bcb..5238878 100644
--- a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/TimeoutKeyedCEPPatternOperator.java
+++ b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/TimeoutKeyedCEPPatternOperator.java
@@ -19,7 +19,6 @@
 package org.apache.flink.cep.operator;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.cep.nfa.NFA;
 import org.apache.flink.cep.nfa.compiler.NFACompiler;
@@ -44,12 +43,11 @@ public class TimeoutKeyedCEPPatternOperator<IN, KEY> extends AbstractKeyedCEPPat
 	public TimeoutKeyedCEPPatternOperator(
 			TypeSerializer<IN> inputSerializer,
 			boolean isProcessingTime,
-			KeySelector<IN, KEY> keySelector,
 			TypeSerializer<KEY> keySerializer,
 			NFACompiler.NFAFactory<IN> nfaFactory,
 			boolean migratingFromOldKeyedOperator) {
 
-		super(inputSerializer, isProcessingTime, keySelector, keySerializer, nfaFactory, migratingFromOldKeyedOperator);
+		super(inputSerializer, isProcessingTime, keySerializer, nfaFactory, migratingFromOldKeyedOperator);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPFrom12MigrationTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPFrom12MigrationTest.java
b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPFrom12MigrationTest.java
index b0f47cc..d9efb1b 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPFrom12MigrationTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPFrom12MigrationTest.java
@@ -78,7 +78,6 @@ public class CEPFrom12MigrationTest {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new NFAFactory(),
 								true),
@@ -126,7 +125,6 @@ public class CEPFrom12MigrationTest {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new NFAFactory(),
 								true),
@@ -200,7 +198,6 @@ public class CEPFrom12MigrationTest {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new NFAFactory(),
 								true),
@@ -246,7 +243,6 @@ public class CEPFrom12MigrationTest {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new NFAFactory(),
 								true),
@@ -332,7 +328,6 @@ public class CEPFrom12MigrationTest {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new SinglePatternNFAFactory(),
 								true),
@@ -371,7 +366,6 @@ public class CEPFrom12MigrationTest {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new SinglePatternNFAFactory(),
 								true),

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigration11to13Test.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigration11to13Test.java
b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigration11to13Test.java
index 8a97448..88a5703 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigration11to13Test.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPMigration11to13Test.java
@@ -103,7 +103,6 @@ public class CEPMigration11to13Test {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								IntSerializer.INSTANCE,
 								new NFAFactory(),
 								true),
@@ -178,7 +177,6 @@ public class CEPMigration11to13Test {
 						new KeyedCEPPatternOperator<>(
 								Event.createTypeSerializer(),
 								false,
-								keySelector,
 								ByteSerializer.INSTANCE,
 								new NFAFactory(),
 								false),

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
index 436ad52..eb50dfd 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPOperatorTest.java
@@ -228,7 +228,6 @@ public class CEPOperatorTest extends TestLogger {
 			new TimeoutKeyedCEPPatternOperator<>(
 				Event.createTypeSerializer(),
 				false,
-				keySelector,
 				IntSerializer.INSTANCE,
 				new NFAFactory(true),
 				true),
@@ -297,7 +296,7 @@ public class CEPOperatorTest extends TestLogger {
 		Event startEventK2 = new Event(43, "start", 1.0);
 
 		TestKeySelector keySelector = new TestKeySelector();
-		KeyedCEPPatternOperator<Event, Integer> operator = getKeyedCepOpearator(false, keySelector);
+		KeyedCEPPatternOperator<Event, Integer> operator = getKeyedCepOpearator(false);
 		OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness
= getCepTestHarness(operator);
 
 		harness.open();
@@ -384,7 +383,6 @@ public class CEPOperatorTest extends TestLogger {
 		KeyedCEPPatternOperator<Event, Integer> operator = new KeyedCEPPatternOperator<>(
 				Event.createTypeSerializer(),
 				false,
-				keySelector,
 				IntSerializer.INSTANCE,
 				new ComplexNFAFactory(),
 				true);
@@ -475,7 +473,7 @@ public class CEPOperatorTest extends TestLogger {
 		Event startEventK2 = new Event(43, "start", 1.0);
 
 		TestKeySelector keySelector = new TestKeySelector();
-		KeyedCEPPatternOperator<Event, Integer> operator = getKeyedCepOpearator(true, keySelector);
+		KeyedCEPPatternOperator<Event, Integer> operator = getKeyedCepOpearator(true);
 		OneInputStreamOperatorTestHarness<Event, Map<String, List<Event>>> harness
= getCepTestHarness(operator);
 
 		harness.open();
@@ -555,7 +553,7 @@ public class CEPOperatorTest extends TestLogger {
 		KeySelector<Event, Integer> keySelector = new TestKeySelector();
 
 		return new KeyedOneInputStreamOperatorTestHarness<>(
-			getKeyedCepOpearator(isProcessingTime, keySelector),
+			getKeyedCepOpearator(isProcessingTime),
 			keySelector,
 			BasicTypeInfo.INT_TYPE_INFO);
 	}
@@ -571,13 +569,11 @@ public class CEPOperatorTest extends TestLogger {
 	}
 
 	private KeyedCEPPatternOperator<Event, Integer> getKeyedCepOpearator(
-			boolean isProcessingTime,
-			KeySelector<Event, Integer> keySelector) {
+			boolean isProcessingTime) {
 
 		return new KeyedCEPPatternOperator<>(
 			Event.createTypeSerializer(),
 			isProcessingTime,
-			keySelector,
 			IntSerializer.INSTANCE,
 			new NFAFactory(),
 			true);

http://git-wip-us.apache.org/repos/asf/flink/blob/34a6020f/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
index 9eb8da2..45d7215 100644
--- a/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
+++ b/flink-libraries/flink-cep/src/test/java/org/apache/flink/cep/operator/CEPRescalingTest.java
@@ -344,7 +344,6 @@ public class CEPRescalingTest {
 			new KeyedCEPPatternOperator<>(
 				Event.createTypeSerializer(),
 				false,
-				keySelector,
 				BasicTypeInfo.INT_TYPE_INFO.createSerializer(new ExecutionConfig()),
 				new NFAFactory(),
 				true),


Mime
View raw message