You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2016/02/01 16:55:51 UTC
[1/2] flink git commit: Revert "[FLINK-3265] [rabbitmq] Fix
concurrency bug in RabbitMQ"
Repository: flink
Updated Branches:
refs/heads/master 5e2fb3cb5 -> 67b380d61
Revert "[FLINK-3265] [rabbitmq] Fix concurrency bug in RabbitMQ"
This reverts commit 6b01a89020f2de3f7710cf72336291b1e8ca8562. The
introduced locks are not necessary. The checkpointing test case only
needs to be adapted to the checkpointing runtime behavior.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/83e6a2b5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/83e6a2b5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/83e6a2b5
Branch: refs/heads/master
Commit: 83e6a2b5f7621a2bfd8797ab071a94492bbac3d5
Parents: 5e2fb3c
Author: Maximilian Michels <mx...@apache.org>
Authored: Mon Feb 1 16:28:07 2016 +0100
Committer: Maximilian Michels <mx...@apache.org>
Committed: Mon Feb 1 16:50:48 2016 +0100
----------------------------------------------------------------------
.../connectors/rabbitmq/RMQSource.java | 4 +-
.../connectors/rabbitmq/RMQSourceTest.java | 79 --------------------
.../source/MessageAcknowledgingSourceBase.java | 51 ++++++-------
...ltipleIdsMessageAcknowledgingSourceBase.java | 24 +++---
4 files changed, 34 insertions(+), 124 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/83e6a2b5/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java b/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
index 59bc057..09bb07c 100644
--- a/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
+++ b/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
@@ -196,9 +196,7 @@ public class RMQSource<OUT> extends MultipleIdsMessageAcknowledgingSourceBase<OU
continue;
}
}
- synchronized (sessionIdsPerSnapshot) {
- sessionIds.add(deliveryTag);
- }
+ sessionIds.add(deliveryTag);
}
ctx.collect(result);
http://git-wip-us.apache.org/repos/asf/flink/blob/83e6a2b5/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java b/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
index 0a3de84..aa19e5d 100644
--- a/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
+++ b/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
@@ -23,7 +23,6 @@ import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.Configuration;
@@ -32,7 +31,6 @@ import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.junit.After;
-import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -105,83 +103,6 @@ public class RMQSourceTest {
sourceThread.join();
}
- /**
- * Make sure concurrent access to snapshotState() and notifyCheckpointComplete() don't cause
- * an issue.
- *
- * Without proper synchronization, the test will fail with a concurrent modification exception
- *
- */
- @Test
- public void testConcurrentAccess() throws Exception {
- source.autoAck = false;
- sourceThread.start();
-
- final Tuple1<Throwable> error = new Tuple1<>(null);
-
- Thread.sleep(5);
-
- Thread snapshotThread = new Thread(new Runnable() {
- public long id = 0;
-
- @Override
- public void run() {
- while (!Thread.interrupted()) {
- try {
- source.snapshotState(id++, 0);
- } catch (Exception e) {
- error.f0 = e;
- break; // stop thread
- }
- }
- }
- });
-
- Thread notifyThread = new Thread(new Runnable() {
- @Override
- public void run() {
- while (!Thread.interrupted()) {
- try {
- // always remove all checkpoints
- source.notifyCheckpointComplete(Long.MAX_VALUE);
- } catch (Exception e) {
- error.f0 = e;
- break; // stop thread
- }
- }
- }
- });
-
- snapshotThread.start();
- notifyThread.start();
-
- long deadline = System.currentTimeMillis() + 1000L;
- while(System.currentTimeMillis() < deadline) {
- if(!snapshotThread.isAlive()) {
- notifyThread.interrupt();
- break;
- }
- if(!notifyThread.isAlive()) {
- snapshotThread.interrupt();
- break;
- }
- Thread.sleep(10);
- }
- if(snapshotThread.isAlive()) {
- snapshotThread.interrupt();
- snapshotThread.join();
- }
- if(notifyThread.isAlive()) {
- notifyThread.interrupt();
- notifyThread.join();
- }
- if(error.f0 != null) {
- error.f0.printStackTrace();
- Assert.fail("Test failed with " + error.f0.getClass().getCanonicalName());
- }
-
- }
-
@Test
public void testCheckpointing() throws Exception {
source.autoAck = false;
http://git-wip-us.apache.org/repos/asf/flink/blob/83e6a2b5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
index 2f865d1..4385884 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MessageAcknowledgingSourceBase.java
@@ -76,7 +76,6 @@ import org.slf4j.LoggerFactory;
* @param <Type> The type of the messages created by the source.
* @param <UId> The type of unique IDs which may be used to acknowledge elements.
*/
-@SuppressWarnings("SynchronizeOnNonFinalField")
public abstract class MessageAcknowledgingSourceBase<Type, UId>
extends RichSourceFunction<Type>
implements Checkpointed<SerializedCheckpointData[]>, CheckpointNotifier {
@@ -167,45 +166,41 @@ public abstract class MessageAcknowledgingSourceBase<Type, UId>
LOG.debug("Snapshotting state. Messages: {}, checkpoint id: {}, timestamp: {}",
idsForCurrentCheckpoint, checkpointId, checkpointTimestamp);
- synchronized (pendingCheckpoints) {
- pendingCheckpoints.addLast(new Tuple2<>(checkpointId, idsForCurrentCheckpoint));
+ pendingCheckpoints.addLast(new Tuple2<>(checkpointId, idsForCurrentCheckpoint));
- idsForCurrentCheckpoint = new ArrayList<>(64);
+ idsForCurrentCheckpoint = new ArrayList<>(64);
- return SerializedCheckpointData.fromDeque(pendingCheckpoints, idSerializer);
- }
+ return SerializedCheckpointData.fromDeque(pendingCheckpoints, idSerializer);
}
@Override
public void restoreState(SerializedCheckpointData[] state) throws Exception {
- synchronized (pendingCheckpoints) {
- pendingCheckpoints = SerializedCheckpointData.toDeque(state, idSerializer);
- // build a set which contains all processed ids. It may be used to check if we have
- // already processed an incoming message.
- for (Tuple2<Long, List<UId>> checkpoint : pendingCheckpoints) {
- idsProcessedButNotAcknowledged.addAll(checkpoint.f1);
- }
+ pendingCheckpoints = SerializedCheckpointData.toDeque(state, idSerializer);
+ // build a set which contains all processed ids. It may be used to check if we have
+ // already processed an incoming message.
+ for (Tuple2<Long, List<UId>> checkpoint : pendingCheckpoints) {
+ idsProcessedButNotAcknowledged.addAll(checkpoint.f1);
}
}
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
LOG.debug("Committing Messages externally for checkpoint {}", checkpointId);
- synchronized (pendingCheckpoints) {
- for (Iterator<Tuple2<Long, List<UId>>> iter = pendingCheckpoints.iterator(); iter.hasNext(); ) {
- Tuple2<Long, List<UId>> checkpoint = iter.next();
- long id = checkpoint.f0;
-
- if (id <= checkpointId) {
- LOG.trace("Committing Messages with following IDs {}", checkpoint.f1);
- acknowledgeIDs(checkpointId, checkpoint.f1);
- // remove deduplication data
- idsProcessedButNotAcknowledged.removeAll(checkpoint.f1);
- // remove checkpoint data
- iter.remove();
- } else {
- break;
- }
+
+ for (Iterator<Tuple2<Long, List<UId>>> iter = pendingCheckpoints.iterator(); iter.hasNext();) {
+ Tuple2<Long, List<UId>> checkpoint = iter.next();
+ long id = checkpoint.f0;
+
+ if (id <= checkpointId) {
+ LOG.trace("Committing Messages with following IDs {}", checkpoint.f1);
+ acknowledgeIDs(checkpointId, checkpoint.f1);
+ // remove deduplication data
+ idsProcessedButNotAcknowledged.removeAll(checkpoint.f1);
+ // remove checkpoint data
+ iter.remove();
+ }
+ else {
+ break;
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/83e6a2b5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
index 4709759..c097066 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/MultipleIdsMessageAcknowledgingSourceBase.java
@@ -107,16 +107,14 @@ public abstract class MultipleIdsMessageAcknowledgingSourceBase<Type, UId, Sessi
*/
protected final void acknowledgeIDs(long checkpointId, List<UId> uniqueIds) {
LOG.debug("Acknowledging ids for checkpoint {}", checkpointId);
- synchronized (sessionIdsPerSnapshot) {
- Iterator<Tuple2<Long, List<SessionId>>> iterator = sessionIdsPerSnapshot.iterator();
- while (iterator.hasNext()) {
- final Tuple2<Long, List<SessionId>> next = iterator.next();
- long id = next.f0;
- if (id <= checkpointId) {
- acknowledgeSessionIDs(next.f1);
- // remove ids for this session
- iterator.remove();
- }
+ Iterator<Tuple2<Long, List<SessionId>>> iterator = sessionIdsPerSnapshot.iterator();
+ while (iterator.hasNext()) {
+ final Tuple2<Long, List<SessionId>> next = iterator.next();
+ long id = next.f0;
+ if (id <= checkpointId) {
+ acknowledgeSessionIDs(next.f1);
+ // remove ids for this session
+ iterator.remove();
}
}
}
@@ -134,10 +132,8 @@ public abstract class MultipleIdsMessageAcknowledgingSourceBase<Type, UId, Sessi
@Override
public SerializedCheckpointData[] snapshotState(long checkpointId, long checkpointTimestamp) throws Exception {
- synchronized (sessionIdsPerSnapshot) {
- sessionIdsPerSnapshot.add(new Tuple2<>(checkpointId, sessionIds));
- sessionIds = new ArrayList<>(64);
- }
+ sessionIdsPerSnapshot.add(new Tuple2<>(checkpointId, sessionIds));
+ sessionIds = new ArrayList<>(64);
return super.snapshotState(checkpointId, checkpointTimestamp);
}
}
[2/2] flink git commit: [FLINK-3265][tests] adapt RMQSource
checkpointing test to runtime behavior
Posted by mx...@apache.org.
[FLINK-3265][tests] adapt RMQSource checkpointing test to runtime behavior
The methods snapshotState and notifyCheckpointComplete are always
mutually exclusive. The RMQSource relies on this but the test makes a
false assumption when it calls those two methods at the same time.
This closes #1569.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/67b380d6
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/67b380d6
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/67b380d6
Branch: refs/heads/master
Commit: 67b380d617a942c11ab29a8d62d67b770245bb63
Parents: 83e6a2b
Author: Maximilian Michels <mx...@apache.org>
Authored: Mon Feb 1 16:30:52 2016 +0100
Committer: Maximilian Michels <mx...@apache.org>
Committed: Mon Feb 1 16:51:35 2016 +0100
----------------------------------------------------------------------
.../flink/streaming/connectors/rabbitmq/RMQSourceTest.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/67b380d6/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java b/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
index aa19e5d..e0eed70 100644
--- a/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
+++ b/flink-streaming-connectors/flink-connector-rabbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceTest.java
@@ -140,7 +140,9 @@ public class RMQSourceTest {
}
// check if the messages are being acknowledged and the transaction comitted
- source.notifyCheckpointComplete(snapshotId);
+ synchronized (DummySourceContext.lock) {
+ source.notifyCheckpointComplete(snapshotId);
+ }
totalNumberOfAcks += numIds;
}