### Question:
PubSubSource extends RichSourceFunction<OUT> implements ResultTypeQueryable<OUT>, ParallelSourceFunction<OUT>, CheckpointListener, ListCheckpointed<AcknowledgeIdsForCheckpoint<String>> { @Override public void open(Configuration configuration) throws Exception { super.open(configuration); if (hasNoCheckpointingEnabled(getRuntimeContext())) { throw new IllegalArgumentException("The PubSubSource REQUIRES Checkpointing to be enabled and " + "the checkpointing frequency must be MUCH lower than the PubSub timeout for it to retry a message."); } getRuntimeContext().getMetricGroup().gauge("PubSubMessagesProcessedNotAcked", this::getOutstandingMessagesToAck); createAndSetPubSubSubscriber(); this.isRunning = true; } PubSubSource(PubSubDeserializationSchema<OUT> deserializationSchema,
				PubSubSubscriberFactory pubSubSubscriberFactory,
				Credentials credentials,
				int maxMessagesToAcknowledge,
				AcknowledgeOnCheckpointFactory acknowledgeOnCheckpointFactory); @Override void open(Configuration configuration); @Override void run(SourceContext<OUT> sourceContext); @Override void cancel(); @Override TypeInformation<OUT> getProducedType(); static DeserializationSchemaBuilder newBuilder(); @Override void notifyCheckpointComplete(long checkpointId); @Override List<AcknowledgeIdsForCheckpoint<String>> snapshotState(long checkpointId, long timestamp); @Override void restoreState(List<AcknowledgeIdsForCheckpoint<String>> state); static final int NO_MAX_MESSAGES_TO_ACKNOWLEDGE_LIMIT; }### Answer:
@Test(expected = IllegalArgumentException.class) public void testOpenWithoutCheckpointing() throws Exception { when(streamingRuntimeContext.isCheckpointingEnabled()).thenReturn(false); pubSubSource.open(null); }
@Test public void testOpenWithCheckpointing() throws Exception { when(streamingRuntimeContext.isCheckpointingEnabled()).thenReturn(true); pubSubSource.open(null); verify(pubSubSubscriberFactory, times(1)).getSubscriber(eq(credentials)); verify(acknowledgeOnCheckpointFactory, times(1)).create(pubsubSubscriber); }