### Question:
RMQSource extends MultipleIdsMessageAcknowledgingSourceBase<OUT, String, Long> implements ResultTypeQueryable<OUT> { @Override public void open(Configuration config) throws Exception { super.open(config); ConnectionFactory factory = setupConnectionFactory(); try { connection = factory.newConnection(); channel = connection.createChannel(); if (channel == null) { throw new RuntimeException("None of RabbitMQ channels are available"); } setupQueue(); consumer = new QueueingConsumer(channel); RuntimeContext runtimeContext = getRuntimeContext(); if (runtimeContext instanceof StreamingRuntimeContext && ((StreamingRuntimeContext) runtimeContext).isCheckpointingEnabled()) { autoAck = false; channel.txSelect(); } else { autoAck = true; } LOG.debug("Starting RabbitMQ source with autoAck status: " + autoAck); channel.basicConsume(queueName, autoAck, consumer); } catch (IOException e) { throw new RuntimeException("Cannot create RMQ connection with " + queueName + " at " + rmqConnectionConfig.getHost(), e); } running = true; } RMQSource(RMQConnectionConfig rmqConnectionConfig, String queueName,
					DeserializationSchema<OUT> deserializationSchema);  RMQSource(RMQConnectionConfig rmqConnectionConfig,
					String queueName, boolean usesCorrelationId, DeserializationSchema<OUT> deserializationSchema); @Override void open(Configuration config); @Override void close(); @Override void run(SourceContext<OUT> ctx); @Override void cancel(); @Override TypeInformation<OUT> getProducedType();  }### Answer:
@Test public void testCheckpointing() throws Exception { source.autoAck = false; StreamSource<String, RMQSource<String>> src = new StreamSource<>(source); AbstractStreamOperatorTestHarness<String> testHarness = new AbstractStreamOperatorTestHarness<>(src, 1, 1, 0); testHarness.open(); sourceThread.start(); Thread.sleep(5); final Random random = new Random(System.currentTimeMillis()); int numSnapshots = 50; long previousSnapshotId; long lastSnapshotId = 0; long totalNumberOfAcks = 0; for (int i = 0; i < numSnapshots; i++) { long snapshotId = random.nextLong(); OperatorSubtaskState data; synchronized (DummySourceContext.lock) { data = testHarness.snapshot(snapshotId, System.currentTimeMillis()); previousSnapshotId = lastSnapshotId; lastSnapshotId = messageId; } Thread.sleep(5); final long numIds = lastSnapshotId - previousSnapshotId; RMQTestSource sourceCopy = new RMQTestSource(); StreamSource<String, RMQTestSource> srcCopy = new StreamSource<>(sourceCopy); AbstractStreamOperatorTestHarness<String> testHarnessCopy = new AbstractStreamOperatorTestHarness<>(srcCopy, 1, 1, 0); testHarnessCopy.setup(); testHarnessCopy.initializeState(data); testHarnessCopy.open(); ArrayDeque<Tuple2<Long, Set<String>>> deque = sourceCopy.getRestoredState(); Set<String> messageIds = deque.getLast().f1; assertEquals(numIds, messageIds.size()); if (messageIds.size() > 0) { assertTrue(messageIds.contains(Long.toString(lastSnapshotId))); } synchronized (DummySourceContext.lock) { source.notifyCheckpointComplete(snapshotId); } totalNumberOfAcks += numIds; } Mockito.verify(source.channel, Mockito.times((int) totalNumberOfAcks)).basicAck(Mockito.anyLong(), Mockito.eq(false)); Mockito.verify(source.channel, Mockito.times(numSnapshots)).txCommit(); }