[flink] Fix read partitioned table throw LogScanner is not subscribed any bucket

[luoyuxia]

Purpose

Linked issue: close #164

Fix read partitioned table throw LogScanner is not subscribed any bucket.

1. In FlinkSourceEnumerator, duplicate the partition of assigned partitions and pendingSplitAssignment to avoid handle the same partitions more than once which will cause one partition still in pending assignment to be removed.
2. In Flink split source reader, when no any subscribed bucket, not call Scanner#poll method, just return empty records directly

FlinkSourceEnumeratorTest#testDiscoverPartitionsPeriodically
FlinkSourceSplitReaderTest#testNoSubscribedBucket

Documentation

N/A