You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2019/01/31 06:36:42 UTC
[flink] branch master updated (3c1bbf2 -> 0e94ecf)
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 3c1bbf2 [hotfix][runtime] Refactor TaskManagerRunner#createRpcService
new 52c2b22 [FLINK-11460] [test] Remove the useless class AcknowledgeStreamMockEnvironment
new 0e94ecf [FLINK-11461] [test] Remove the useless MockRecordReader class
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/operators/sort/MockRecordReader.java | 105 ---------------------
.../tasks/AcknowledgeStreamMockEnvironment.java | 77 ---------------
2 files changed, 182 deletions(-)
delete mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MockRecordReader.java
delete mode 100644 flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AcknowledgeStreamMockEnvironment.java
[flink] 02/02: [FLINK-11461] [test] Remove the useless
MockRecordReader class
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0e94ecfc7b65021905087b7efa94e3e04f761baf
Author: Zhijiang <wa...@aliyun.com>
AuthorDate: Thu Jan 31 10:59:51 2019 +0800
[FLINK-11461] [test] Remove the useless MockRecordReader class
This closes #7610.
---
.../runtime/operators/sort/MockRecordReader.java | 105 ---------------------
1 file changed, 105 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MockRecordReader.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MockRecordReader.java
deleted file mode 100644
index e5b8ed0..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/MockRecordReader.java
+++ /dev/null
@@ -1,105 +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.runtime.operators.sort;
-
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-
-import org.apache.flink.types.Record;
-import org.apache.flink.util.MutableObjectIterator;
-
-/**
- */
-public class MockRecordReader implements MutableObjectIterator<Record> {
- private final Record SENTINEL = new Record();
-
- private final BlockingQueue<Record> queue;
-
- public MockRecordReader() {
- this.queue = new ArrayBlockingQueue<Record>(32, false);
- }
-
- public MockRecordReader(int size) {
- this.queue = new ArrayBlockingQueue<Record>(size, false);
- }
-
- @Override
- public Record next(Record reuse) {
- Record r = null;
- while (r == null) {
- try {
- r = queue.take();
- } catch (InterruptedException iex) {
- throw new RuntimeException("Reader was interrupted.");
- }
- }
-
- if (r == SENTINEL) {
- // put the sentinel back, to ensure that repeated calls do not block
- try {
- queue.put(r);
- } catch (InterruptedException e) {
- throw new RuntimeException("Reader was interrupted.");
- }
- return null;
- } else {
- r.copyTo(reuse);
- return reuse;
- }
- }
-
- @Override
- public Record next() {
- Record r = null;
- while (r == null) {
- try {
- r = queue.take();
- } catch (InterruptedException iex) {
- throw new RuntimeException("Reader was interrupted.");
- }
- }
-
- if (r == SENTINEL) {
- // put the sentinel back, to ensure that repeated calls do not block
- try {
- queue.put(r);
- } catch (InterruptedException e) {
- throw new RuntimeException("Reader was interrupted.");
- }
- return null;
- } else {
- Record result = new Record(r.getNumFields());
- r.copyTo(result);
- return result;
- }
- }
-
- public void emit(Record element) throws InterruptedException {
- queue.put(element.createCopy());
- }
-
- public void close() {
- try {
- queue.put(SENTINEL);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
- }
-}
[flink] 01/02: [FLINK-11460] [test] Remove the useless class
AcknowledgeStreamMockEnvironment
Posted by tz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 52c2b22bd047912c690edd1d770d567250f6d710
Author: Zhijiang <wa...@aliyun.com>
AuthorDate: Thu Jan 31 11:07:56 2019 +0800
[FLINK-11460] [test] Remove the useless class AcknowledgeStreamMockEnvironment
This closes #7611.
---
.../tasks/AcknowledgeStreamMockEnvironment.java | 77 ----------------------
1 file changed, 77 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AcknowledgeStreamMockEnvironment.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AcknowledgeStreamMockEnvironment.java
deleted file mode 100644
index 8941cc1..0000000
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/AcknowledgeStreamMockEnvironment.java
+++ /dev/null
@@ -1,77 +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.streaming.runtime.tasks;
-
-import org.apache.flink.api.common.ExecutionConfig;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.core.testutils.OneShotLatch;
-import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
-import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
-import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider;
-import org.apache.flink.runtime.state.TaskStateManager;
-
-/**
- * Stream environment that allows to wait for checkpoint acknowledgement.
- */
-public class AcknowledgeStreamMockEnvironment extends StreamMockEnvironment {
- private final OneShotLatch checkpointLatch = new OneShotLatch();
- private volatile long checkpointId;
- private volatile TaskStateSnapshot checkpointStateHandles;
-
- public AcknowledgeStreamMockEnvironment(
- Configuration jobConfig,
- Configuration taskConfig,
- ExecutionConfig executionConfig,
- long memorySize,
- MockInputSplitProvider inputSplitProvider,
- int bufferSize,
- TaskStateManager taskStateManager) {
- super(
- jobConfig,
- taskConfig,
- executionConfig,
- memorySize,
- inputSplitProvider,
- bufferSize,
- taskStateManager);
- }
-
- public long getCheckpointId() {
- return checkpointId;
- }
-
- @Override
- public void acknowledgeCheckpoint(
- long checkpointId,
- CheckpointMetrics checkpointMetrics,
- TaskStateSnapshot checkpointStateHandles) {
-
- this.checkpointId = checkpointId;
- this.checkpointStateHandles = checkpointStateHandles;
- checkpointLatch.trigger();
- }
-
- public OneShotLatch getCheckpointLatch() {
- return checkpointLatch;
- }
-
- public TaskStateSnapshot getCheckpointStateHandles() {
- return checkpointStateHandles;
- }
-}