Add validation to check for non-ints from partitionfns

sdks/python/apache_beam/transforms/core.py

@@ -3577,6 +3577,10 @@ def process(self, element, partitionfn, n, *args, **kwargs):
         raise ValueError(
             'PartitionFn specified out-of-bounds partition index: '
             '%d not in [0, %d)' % (partition, n))
+      if isinstance(partition, bool) or not isinstance(partition, int):
+        raise ValueError(
+            f"PartitionFn yielded a '{type(partition).__name__}' "
+            "when it should only yields integers")
       # Each input is directed into the output that corresponds to the
       # selected partition.
       yield pvalue.TaggedOutput(str(partition), element)

sdks/python/apache_beam/transforms/core_test.py

@@ -162,6 +162,14 @@ def test_dofn_with_implicit_return_none_return_without_value(self):
 
 
 class PartitionTest(unittest.TestCase):
+  def test_partition_with_bools(self):
+    with pytest.raises(
+        ValueError,
+        match="PartitionFn yielded a 'bool' when it should only yields integers"
+    ):
+      with beam.testing.test_pipeline.TestPipeline() as p:
+        _ = (p | beam.Create([True]) | beam.Partition(lambda x, _: x, 2))
+
   def test_partition_boundedness(self):
     def partition_fn(val, num_partitions):
       return val % num_partitions