[FLINK-3301] Ineffective synchronization in MessageAcknowledgingSourceBase#restoreState

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Duplicate
• Assignee: Unassigned
• Reporter: Ted Yu
• Created: 28/Jan/16 22:13
• Updated: 01/Feb/16 15:39
• Resolved: 01/Feb/16 11:36

Description

Here is related code:

  public void restoreState(SerializedCheckpointData[] state) throws Exception {
    synchronized (pendingCheckpoints) {
      pendingCheckpoints = SerializedCheckpointData.toDeque(state, idSerializer);

When lock is held on current pendingCheckpoints field, new reference is assigned to the same field - making the synchronization ineffective.

Issue Links

• duplicates: FLINK-3265 RabbitMQ Source not threadsafe: ConcurrentModificationException
• relates to: FLINK-3307 RMQSourceTest failure