[Bug]: Python Nexmark Query 1 is broken unsupported operand & float and int

### What happened?

```
./gradlew --continue -Dorg.gradle.jvmargs=-Xms2g -Dorg.gradle.jvmargs=-Xmx6g -Dorg.gradle.vfs.watch=false -Pdocker-pull-licenses -Pnexmark.runner=:runners:direct-java "-Pnexmark.args=--manageResources=false --monitorJobs=true --bigQueryTable=nexmark --bigQueryDataset=nexmark_PRs --project=apache-beam-testing --resourceNameMode=QUERY_RUNNER_AND_MODE --exportSummaryToBigQuery=true --tempLocation=gs://temp-storage-for-perf-tests/nexmark --influxDatabase=beam_test_metrics --influxHost=http://10.128.0.96:8086/ --baseInfluxMeasurement=nexmark_PRs --exportSummaryToInfluxDB=true --influxRetentionPolicy=forever --suite=SMOKE --enforceEncodability=true --enforceImmutability=true --runner=DirectRunner --query=1 --numEvents=100000 --generateEventFilePathPrefix=gs://temp-storage-for-perf-tests/nexmark/eventFiles/jenkins-beam_PostCommit_Python_Nexmark_Direct_PR-5/query1-" :sdks:java:testing:nexmark:run

./gradlew --continue -Dorg.gradle.jvmargs=-Xms2g -Dorg.gradle.jvmargs=-Xmx6g -Dorg.gradle.vfs.watch=false -Pdocker-pull-licenses "-Pnexmark.args= --monitor_jobs --big_query_table=nexmark --big_query_dataset=nexmark_PRs --project=apache-beam-testing --resource_name_mode=QUERY_RUNNER_AND_MODE --export_summary_to_big_query --temp_location=gs://temp-storage-for-perf-tests/nexmark --influx_database=beam_test_metrics --influx_host=http://10.128.0.96:8086/ --base_influx_measurement=nexmark_PRs --export_summary_to_influx_db --influx_retention_policy=forever --suite=SMOKE --enforce_encodability --enforce_immutability --runner=DirectRunner --query=1 --num_events=100000 --input=gs://temp-storage-for-perf-tests/nexmark/eventFiles/jenkins-beam_PostCommit_Python_Nexmark_Direct_PR-5/query1-\*" :sdks:python:apache_beam:testing:benchmarks:nexmark:run
```

for environment ref_Environment_default_environment_1 (beam:env:embedded_python:v1, b'')
21:02:40 (MainThread) Finished listing 1 files in 0.0431971549987793 seconds.
21:02:40 Traceback (most recent call last):
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1417, in process
21:02:41 return self.do_fn_invoker.invoke_process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 623, in invoke_process
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 239, in receive
21:02:41 self.update_counters_start(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 198, in update_counters_start
21:02:41 self.opcounter.update_from(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/opcounters.py", line 215, in update_from
21:02:41 self.do_sample(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/opcounters.py", line 267, in do_sample
21:02:41 self.coder_impl.get_estimated_size_and_observables(windowed_value))
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 1490, in get_estimated_size_and_observables
21:02:41 self._value_coder.get_estimated_size_and_observables(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 209, in get_estimated_size_and_observables
21:02:41 return self.estimate_size(value, nested), []
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 248, in estimate_size
21:02:41 self.encode_to_stream(value, out, nested)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/models/nexmark_model.py", line 182, in encode_to_stream
21:02:41 self._int_coder_impl.encode_to_stream(value.price, stream, True)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 936, in encode_to_stream
21:02:41 out.write_var_int64(value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/slow_stream.py", line 55, in write_var_int64
21:02:41 bits = v & 0x7F
21:02:41 TypeError: unsupported operand type(s) for &: 'float' and 'int'
21:02:41
21:02:41 During handling of the above exception, another exception occurred:
21:02:41
21:02:41 Traceback (most recent call last):
21:02:41 File "/usr/lib/python3.8/runpy.py", line 194, in _run_module_as_main
21:02:41 return _run_code(code, main_globals, None,
21:02:41 File "/usr/lib/python3.8/runpy.py", line 87, in _run_code
21:02:41 exec(code, run_globals)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/nexmark_launcher.py", line 528, in
21:02:41 launcher.run()
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/nexmark_launcher.py", line 513, in run
21:02:41 self.run_query(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/nexmark_launcher.py", line 306, in run_query
21:02:41 result = self.pipeline.run()
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/pipeline.py", line 577, in run
21:02:41 return self.runner.run_pipeline(self, self._options)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/direct/direct_runner.py", line 131, in run_pipeline
21:02:41 return runner.run_pipeline(pipeline, options)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 201, in run_pipeline
21:02:41 self._latest_run_result = self.run_via_runner_api(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 222, in run_via_runner_api
21:02:41 return self.run_stages(stage_context, stages)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 453, in run_stages
21:02:41 bundle_results = self._execute_bundle(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 781, in _execute_bundle
21:02:41 self._run_bundle(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1010, in _run_bundle
21:02:41 result, splits = bundle_manager.process_bundle(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 1346, in process_bundle
21:02:41 result_future = self._****_handler.control_conn.push(process_bundle_req)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/portability/fn_api_runner/****_handlers.py", line 379, in push
21:02:41 response = self.****.do_instruction(request)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/sdk_****.py", line 596, in do_instruction
21:02:41 return getattr(self, request_type)(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/sdk_****.py", line 634, in process_bundle
21:02:41 bundle_processor.process_bundle(instruction_id))
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/bundle_processor.py", line 1003, in process_bundle
21:02:41 input_op_by_transform_id[element.transform_id].process_encoded(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/bundle_processor.py", line 227, in process_encoded
21:02:41 self.output(decoded_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 528, in output
21:02:41 _cast_to_receiver(self.receivers[output_index]).receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 240, in receive
21:02:41 self.consumer.process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 1030, in process
21:02:41 delayed_applications = self.dofn_runner.process_with_sized_restriction(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1432, in process_with_sized_restriction
21:02:41 return self.do_fn_invoker.invoke_process(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 817, in invoke_process
21:02:41 residual = self._invoke_process_per_window(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 981, in _invoke_process_per_window
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 240, in receive
21:02:41 self.consumer.process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 908, in process
21:02:41 delayed_applications = self.dofn_runner.process(o)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1419, in process
21:02:41 self._reraise_augmented(exn)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1491, in _reraise_augmented
21:02:41 raise exn
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1417, in process
21:02:41 return self.do_fn_invoker.invoke_process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 623, in invoke_process
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 240, in receive
21:02:41 self.consumer.process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 908, in process
21:02:41 delayed_applications = self.dofn_runner.process(o)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1419, in process
21:02:41 self._reraise_augmented(exn)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1491, in _reraise_augmented
21:02:41 raise exn
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1417, in process
21:02:41 return self.do_fn_invoker.invoke_process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 623, in invoke_process
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 240, in receive
21:02:41 self.consumer.process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 908, in process
21:02:41 delayed_applications = self.dofn_runner.process(o)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1419, in process
21:02:41 self._reraise_augmented(exn)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1491, in _reraise_augmented
21:02:41 raise exn
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1417, in process
21:02:41 return self.do_fn_invoker.invoke_process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 837, in invoke_process
21:02:41 self._invoke_process_per_window(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 981, in _invoke_process_per_window
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 240, in receive
21:02:41 self.consumer.process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 908, in process
21:02:41 delayed_applications = self.dofn_runner.process(o)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1419, in process
21:02:41 self._reraise_augmented(exn)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1491, in _reraise_augmented
21:02:41 raise exn
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1417, in process
21:02:41 return self.do_fn_invoker.invoke_process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 623, in invoke_process
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 240, in receive
21:02:41 self.consumer.process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 908, in process
21:02:41 delayed_applications = self.dofn_runner.process(o)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1419, in process
21:02:41 self._reraise_augmented(exn)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1507, in _reraise_augmented
21:02:41 raise new_exn.with_traceback(tb)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1417, in process
21:02:41 return self.do_fn_invoker.invoke_process(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 623, in invoke_process
21:02:41 self.output_handler.handle_process_outputs(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1581, in handle_process_outputs
21:02:41 self._write_value_to_tag(tag, windowed_value, watermark_estimator)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/common.py", line 1694, in _write_value_to_tag
21:02:41 self.main_receivers.receive(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 239, in receive
21:02:41 self.update_counters_start(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/operations.py", line 198, in update_counters_start
21:02:41 self.opcounter.update_from(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/opcounters.py", line 215, in update_from
21:02:41 self.do_sample(windowed_value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/runners/****/opcounters.py", line 267, in do_sample
21:02:41 self.coder_impl.get_estimated_size_and_observables(windowed_value))
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 1490, in get_estimated_size_and_observables
21:02:41 self._value_coder.get_estimated_size_and_observables(
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 209, in get_estimated_size_and_observables
21:02:41 return self.estimate_size(value, nested), []
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 248, in estimate_size
21:02:41 self.encode_to_stream(value, out, nested)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/testing/benchmarks/nexmark/models/nexmark_model.py", line 182, in encode_to_stream
21:02:41 self._int_coder_impl.encode_to_stream(value.price, stream, True)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/coder_impl.py", line 936, in encode_to_stream
21:02:41 out.write_var_int64(value)
21:02:41 File "/home/jenkins/jenkins-slave/workspace/beam_PostCommit_Python_Nexmark_Direct_PR/src/sdks/python/apache_beam/coders/slow_stream.py", line 55, in write_var_int64
21:02:41 bits = v & 0x7F
21:02:41 TypeError: unsupported operand type(s) for &: 'float' and 'int' [while running 'ConvertToEuro']

### Issue Priority

Priority: 3 (minor)

### Issue Components

- [X] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner