[BEAM-1780] BigtableIO: better handling of bad split requests

The contract for `splitIntoFraction` is that it should only throw if the
reader is in an unknown, bad state. The proper way to reject invalid or
unsatisfiable split requests is to return null.

However, `BigtableIO.Read` will currently throw for simply invalid input
it should reject. This can lead to less effective dynamic work
rebalancing and even stuck jobs.

Related to, but probably not a complete solution for, BEAM-1751.

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java

@@ -998,19 +998,32 @@ public final long getSplitPointsConsumed() {
  }
 
  @Override
+ @Nullable
  public final synchronized BigtableSource splitAtFraction(double fraction) {
  ByteKey splitKey;
  try {
  splitKey = rangeTracker.getRange().interpolateKey(fraction);
- } catch (IllegalArgumentException e) {
+ } catch (RuntimeException e) {
  LOG.info(
- "%s: Failed to interpolate key for fraction %s.", rangeTracker.getRange(), fraction);
+ "%s: Failed to interpolate key for fraction %s.", rangeTracker.getRange(), fraction, e);
  return null;
  }
  LOG.debug(
  "Proposing to split {} at fraction {} (key {})", rangeTracker, fraction, splitKey);
- BigtableSource primary = source.withEndKey(splitKey);
- BigtableSource residual = source.withStartKey(splitKey);
+ BigtableSource primary;
+ BigtableSource residual;
+ try {
+ primary = source.withEndKey(splitKey);
+ residual = source.withStartKey(splitKey);
+ } catch (RuntimeException e) {
+ LOG.info(
+ "%s: Interpolating for fraction %s yielded invalid split key %s.",
+ rangeTracker.getRange(),
+ fraction,
+ splitKey,
+ e);
+ return null;
+ }
  if (!rangeTracker.trySplitAtPosition(splitKey)) {
  return null;
  }