fix the error when num of partitions increased

[tilumi]

Dedicated EventHub support partition increase, but Spark job get failed due to checkpoint doesn't contain sequence number for new partitions:

Caused by: java.lang.IllegalStateException: {"ehName":"hygieneraw","partitionId":63} doesn't have a fromSeqNo
at org.apache.spark.sql.eventhubs.EventHubsSource$$anonfun$7$$anonfun$8.apply(EventHubsSource.scala:283)
at org.apache.spark.sql.eventhubs.EventHubsSource$$anonfun$7$$anonfun$8.apply(EventHubsSource.scala:283)
at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
at scala.collection.AbstractMap.getOrElse(Map.scala:59)
at org.apache.spark.sql.eventhubs.EventHubsSource$$anonfun$7.apply(EventHubsSource.scala:281)
at org.apache.spark.sql.eventhubs.EventHubsSource$$anonfun$7.apply(EventHubsSource.scala:280)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)

This PR contains following change:

1. get num of partitions for every batch
2. introduce parameter "StartingPositionForNewPartitions", which specify starting position for new partitions, default is fromStartOfStream
3. In EventHubsSource, check if num of partitions increased, if yes, translate starting positions to sequence numbers for new partitions, combined with sequence numbers of previous batch as sequence numbers for new batch

[nyaghma]

The proposed change updates the common path of a batch execution by getting the number of partitions for every single batch (partitionCount is not a lazy val anymore and this affects the execution of earliestAndLatest for every batch).
This is an overhead added to every batch execution for a scenario that would happen rarely. Therefore, instead of changing the common case scenario in batch execution we can add an exception handler which handles this scenario when needed.

[tilumi]

The proposed change updates the common path of a batch execution by getting the number of partitions for every single batch (partitionCount is not a lazy val anymore and this affects the execution of earliestAndLatest for every batch).
This is an overhead added to every batch execution for a scenario that would happen rarely. Therefore, instead of changing the common case scenario in batch execution we can add an exception handler which handles this scenario when needed.

I remove getting number of partitions for each batch part, then users should restart their jobs every time partition number increased

[nyaghma]

Could you add an exception handling scenario so that when the exception happens it gets the new partition size and adds the new partitions. In this way, the user doesn't have to restart when he/she adds a new partition.

[sjkwak]

Hi @tilumi - have we done an upgrade testing with the change and verify whether the latest iteration works as expected?