diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index d9f3de5ea766..c8ce008a09e3 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -1579,7 +1579,6 @@ class BeamModulePlugin implements Plugin { "DirectInvocationOnMock", "Finalize", "JUnitIncompatibleType", - "LongDoubleConversion", "MockNotUsedInProduction", "NarrowCalculation", "NullableTypeParameter", diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java index 7ea8dece31bb..17ae820f1c13 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/bigquery/BigQueryIOLT.java @@ -341,7 +341,7 @@ private void testRead() throws IOException { region, launchInfo.jobId(), getBeamMetricsName(PipelineMetricsType.COUNTER, READ_ELEMENT_METRIC_NAME)); - assertEquals(configuration.numRecords, numRecords, 0.5); + assertEquals((double) configuration.numRecords, numRecords, 0.5); // export metrics MetricsConfiguration metricsConfig = diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java index 949b863be3df..e2ef7ba33bdf 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/spanner/SpannerIOLT.java @@ -156,7 +156,7 @@ public void testSpannerWriteAndRead() throws IOException { region, readInfo.jobId(), getBeamMetricsName(PipelineMetricsType.COUNTER, READ_ELEMENT_METRIC_NAME)); - assertEquals(configuration.numRecords, numRecords, 0.5); + assertEquals((double) configuration.numRecords, numRecords, 0.5); // export metrics MetricsConfiguration metricsConfig = diff --git a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java index 82caea312189..2da8c0dc134f 100644 --- a/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java +++ b/it/google-cloud-platform/src/test/java/org/apache/beam/it/gcp/storage/FileBasedIOLT.java @@ -221,7 +221,7 @@ public void testTextIOWriteThenRead() throws IOException { readInfo.jobId(), getBeamMetricsName(PipelineMetricsType.COUNTER, READ_ELEMENT_METRIC_NAME)); - assertEquals(configuration.numRecords, numRecords, 0.5); + assertEquals((double) configuration.numRecords, numRecords, 0.5); // export metrics MetricsConfiguration metricsConfig = diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WriteWithShardingFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WriteWithShardingFactory.java index e976180ae190..7f779c21b050 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WriteWithShardingFactory.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WriteWithShardingFactory.java @@ -138,7 +138,7 @@ private int calculateShards(long totalRecords) { return (int) totalRecords; } // 100mil records before >7 output files - int floorLogRecs = (int) Math.log10(totalRecords); + int floorLogRecs = (int) Math.log10((double) totalRecords); return Math.max(floorLogRecs, MIN_SHARDS_FOR_LOG) + extraShards; } } diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSources.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSources.java index 0181e647ac7b..86d6836dfda2 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSources.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSources.java @@ -661,7 +661,7 @@ public void close() throws IOException { @VisibleForTesting static @Nullable ReportedParallelism longToParallelism(long value) { if (value >= 0) { - return new ReportedParallelism().setValue(Double.valueOf(value)); + return new ReportedParallelism().setValue(Double.valueOf((double) value)); } else { return null; } diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/counters/CounterFactory.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/counters/CounterFactory.java index f75fcad6cc4b..25a3083a314b 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/counters/CounterFactory.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/counters/CounterFactory.java @@ -226,7 +226,7 @@ public final CounterDistribution addValue(long value) { long count = this.getCount() + 1; long sum = this.getSum() + value; // TODO: Replace sum-of-squares with statistics for a better stddev algorithm. - double sumOfSquares = this.getSumOfSquares() + Math.pow(value, 2); + double sumOfSquares = this.getSumOfSquares() + Math.pow((double) value, 2); int bucketIndex = calculateBucket(value); List buckets = incrementBucket(bucketIndex); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java index 8b0e4ee433fa..43920abb7371 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java @@ -298,7 +298,7 @@ public void splitRestriction( desiredChunkSize = 64 << 20; // 64mb } else { // 1mb --> 1 shard; 1gb --> 32 shards; 1tb --> 1000 shards, 1pb --> 32k shards - desiredChunkSize = Math.max(1 << 20, (long) (1000 * Math.sqrt(estimatedSize))); + desiredChunkSize = Math.max(1 << 20, (long) (1000 * Math.sqrt((double) estimatedSize))); } List splits = (List) restriction.split(desiredChunkSize, pipelineOptions); @@ -1079,7 +1079,7 @@ private Progress tryGetProgressOrThrow() throws IOException { if (size != UnboundedReader.BACKLOG_UNKNOWN) { // The UnboundedSource/UnboundedReader API has no way of reporting how much work // has been completed so runners can only see the work remaining changing. - return Progress.from(0, size); + return Progress.from(0, (double) size); } // TODO: Support "global" backlog reporting diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java index 5cf9276c766a..5d15d23afe0f 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateQuantiles.java @@ -277,7 +277,7 @@ public ApproximateQuantilesCombineFn withEpsilon(double epsilon) * of {@code maxNumElements}. */ public ApproximateQuantilesCombineFn withMaxInputSize(long maxNumElements) { - return create(numQuantiles, compareFn, maxNumElements, maxNumElements); + return create(numQuantiles, compareFn, maxNumElements, (double) maxNumElements); } /** @@ -484,7 +484,8 @@ private QuantileBuffer collapse(Iterable> buffers) { newLevel = Math.max(newLevel, buffer.level + 1); newWeight += buffer.weight; } - List newElements = interpolate(buffers, bufferSize, newWeight, offset(newWeight)); + List newElements = + interpolate(buffers, bufferSize, (double) newWeight, (double) offset(newWeight)); return new QuantileBuffer<>(newLevel, newWeight, newElements); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java index ce10885b0686..6b511736a733 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java @@ -234,7 +234,7 @@ public long nextBackOffMillis() { Math.min( backoffConfig.initialBackoff.getMillis() * Math.pow(backoffConfig.exponent, currentRetry), - backoffConfig.maxBackoff.getMillis()); + (double) backoffConfig.maxBackoff.getMillis()); double randomOffset = (Math.random() * 2 - 1) * DEFAULT_RANDOMIZATION_FACTOR * currentIntervalMillis; long nextBackoffMillis = Math.round(currentIntervalMillis + randomOffset); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java index 21b7f1ab415e..7cd820c1a389 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/splittabledofn/GrowableOffsetRangeTrackerTest.java @@ -183,7 +183,7 @@ public void testProgressAfterFinished() throws Exception { tracker.checkDone(); simpleEstimator.setEstimateRangeEnd(0L); Progress currentProgress = tracker.getProgress(); - assertEquals(Long.MAX_VALUE - 10L, currentProgress.getWorkCompleted(), 0); + assertEquals((double) (Long.MAX_VALUE - 10L), currentProgress.getWorkCompleted(), 0); assertEquals(0, currentProgress.getWorkRemaining(), 0.001); } @@ -197,23 +197,23 @@ public void testProgress() throws Exception { simpleEstimator.setEstimateRangeEnd(5L); Progress currentProgress = tracker.getProgress(); - assertEquals(cur - start, currentProgress.getWorkCompleted(), 0.001); + assertEquals((double) (cur - start), currentProgress.getWorkCompleted(), 0.001); assertEquals(0, currentProgress.getWorkRemaining(), 0.001); simpleEstimator.setEstimateRangeEnd(35L); currentProgress = tracker.getProgress(); - assertEquals(cur - start, currentProgress.getWorkCompleted(), 0.001); - assertEquals(35L - cur, currentProgress.getWorkRemaining(), 0.001); + assertEquals((double) (cur - start), currentProgress.getWorkCompleted(), 0.001); + assertEquals((double) (35L - cur), currentProgress.getWorkRemaining(), 0.001); simpleEstimator.setEstimateRangeEnd(25L); currentProgress = tracker.getProgress(); - assertEquals(cur - start, currentProgress.getWorkCompleted(), 0.001); - assertEquals(25L - cur, currentProgress.getWorkRemaining(), 0.001); + assertEquals((double) (cur - start), currentProgress.getWorkCompleted(), 0.001); + assertEquals((double) (25L - cur), currentProgress.getWorkRemaining(), 0.001); simpleEstimator.setEstimateRangeEnd(Long.MAX_VALUE); currentProgress = tracker.getProgress(); - assertEquals(cur - start, currentProgress.getWorkCompleted(), 0.001); - assertEquals(Long.MAX_VALUE - cur, currentProgress.getWorkRemaining(), 0.001); + assertEquals((double) (cur - start), currentProgress.getWorkCompleted(), 0.001); + assertEquals((double) (Long.MAX_VALUE - cur), currentProgress.getWorkRemaining(), 0.001); } @Test diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/LatencyRecordingHttpRequestInitializer.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/LatencyRecordingHttpRequestInitializer.java index 5b9cc614e9f2..dcd17f956a63 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/LatencyRecordingHttpRequestInitializer.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/LatencyRecordingHttpRequestInitializer.java @@ -63,7 +63,7 @@ public LoggingInterceptor(Histogram requestLatencies) { @Override public void interceptResponse(HttpResponse response) throws IOException { long timeToResponse = System.currentTimeMillis() - startTime; - requestLatencies.update(timeToResponse); + requestLatencies.update((double) timeToResponse); } @Override diff --git a/sdks/java/extensions/sql/perf-tests/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BigQueryIOPushDownIT.java b/sdks/java/extensions/sql/perf-tests/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BigQueryIOPushDownIT.java index 224dc1ad92c1..c1fd5ad7b4c8 100644 --- a/sdks/java/extensions/sql/perf-tests/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BigQueryIOPushDownIT.java +++ b/sdks/java/extensions/sql/perf-tests/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/bigquery/BigQueryIOPushDownIT.java @@ -199,7 +199,7 @@ private Set> getReadSuppliers( suppliers.add( reader -> { long fieldsRead = reader.getCounterMetric(FIELDS_READ_METRIC); - return NamedTestResult.create(uuid, timestamp, FIELDS_READ_METRIC, fieldsRead); + return NamedTestResult.create(uuid, timestamp, FIELDS_READ_METRIC, (double) fieldsRead); }); return suppliers; } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java index 7c0ab785ae7e..6d5b8f2a9e03 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java @@ -1003,7 +1003,8 @@ public void processElement( private long getDesiredChunkSize( PipelineOptions options, BigQueryStorageTableSource output) throws Exception { - return Math.max(1 << 20, (long) (1000 * Math.sqrt(output.getEstimatedSizeBytes(options)))); + return Math.max( + 1 << 20, (long) (1000 * Math.sqrt((double) output.getEstimatedSizeBytes(options)))); } } @@ -1755,7 +1756,8 @@ private PCollection expandForDirectRead( desiredChunkSize = 64 << 20; // 64mb } else { // 1mb --> 1 shard; 1gb --> 32 shards; 1tb --> 1000 shards, 1pb --> 32k shards - desiredChunkSize = Math.max(1 << 20, (long) (1000 * Math.sqrt(estimatedSize))); + desiredChunkSize = + Math.max(1 << 20, (long) (1000 * Math.sqrt((double) estimatedSize))); } sources = source.split(desiredChunkSize, bqOptions); } catch (Exception e) { diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java index 23da3723045c..f7eed420fdd2 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java @@ -139,7 +139,7 @@ private static void updateRpcLatencyMetric(@Nonnull Context c, RpcMethod meth } long timeElapsed = java.time.Duration.between(operationStartTime, operationEndTime).toMillis(); if (timeElapsed > 0) { - BigQuerySinkMetrics.createRPCLatencyHistogram(method).update(timeElapsed); + BigQuerySinkMetrics.createRPCLatencyHistogram(method).update((double) timeElapsed); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingInsertsMetrics.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingInsertsMetrics.java index 2ada2e94dc12..c372fa801151 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingInsertsMetrics.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingInsertsMetrics.java @@ -132,7 +132,7 @@ public void updateSuccessfulRpcMetrics(Instant start, Instant end) { /** Record rpc latency histogram metrics. */ private void recordRpcLatencyMetrics() { for (Duration d : rpcLatencies()) { - LATENCY_HISTOGRAM.update(d.toMillis()); + LATENCY_HISTOGRAM.update((double) d.toMillis()); } } diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java index b51442a0955f..978b47e83b0c 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/firestore/RpcQosImpl.java @@ -653,7 +653,7 @@ private double calcMaxRequestBudget(Instant instant, Instant first) { long calculatedGrowth = (durationSinceFirst.getStandardMinutes() - rampUpIntervalMinutes) / rampUpIntervalMinutes; long growth = Math.max(0, calculatedGrowth); - return baseBatchBudget * Math.pow(1.5, growth); + return baseBatchBudget * Math.pow(1.5, (double) growth); } void recordWriteCount(Instant instant, int numWrites) { @@ -801,7 +801,8 @@ private BackoffResult doBackoff() { double currentIntervalMillis = Math.min( - initialBackoff.getMillis() * Math.pow(1.5, attempt - 1), MAX_BACKOFF.getMillis()); + initialBackoff.getMillis() * Math.pow(1.5, attempt - 1), + (double) MAX_BACKOFF.getMillis()); double randomOffset = (rand.nextDouble() * 2 - 1) * RANDOMIZATION_FACTOR * currentIntervalMillis; long nextBackoffMillis = Math.round(currentIntervalMillis + randomOffset); diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java index 582d9e709c9a..49540e20594a 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java @@ -153,7 +153,7 @@ public void testProcessElementAndGetSize() throws IOException, InterruptedExcept testToken, tenSecondsAgo, BigDecimal.valueOf(20), Instant.now(), false)); // we should have output 2 100B mutations in the past 10s long bytesPerSecond = (mutationSize * 2) / 10; - assertEquals(sizeEstimate, bytesPerSecond * watermarkLag, 10); + assertEquals(sizeEstimate, (double) (bytesPerSecond * watermarkLag), 10); verify(receiver, times(2)).outputWithTimestamp(KV.of(rowKey, mockMutation), Instant.EPOCH); } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/estimator/NullSizeEstimatorTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/estimator/NullSizeEstimatorTest.java index a229c0e66bef..7fd93b17d62b 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/estimator/NullSizeEstimatorTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigtable/changestreams/estimator/NullSizeEstimatorTest.java @@ -27,8 +27,8 @@ public class NullSizeEstimatorTest { @Test public void alwaysReturns0AsEstimatedThroughput() { final NullSizeEstimator estimator = new NullSizeEstimator<>(); - assertEquals(estimator.sizeOf(new byte[40]), 0D, DELTA); - assertEquals(estimator.sizeOf(new byte[20]), 0D, DELTA); - assertEquals(estimator.sizeOf(new byte[10]), 0D, DELTA); + assertEquals((double) estimator.sizeOf(new byte[40]), 0D, DELTA); + assertEquals((double) estimator.sizeOf(new byte[20]), 0D, DELTA); + assertEquals((double) estimator.sizeOf(new byte[10]), 0D, DELTA); } } diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/TimestampRangeTrackerTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/TimestampRangeTrackerTest.java index b57acab8e3ec..384ada6cebb5 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/TimestampRangeTrackerTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/restriction/TimestampRangeTrackerTest.java @@ -236,8 +236,9 @@ public void testGetProgressWorkCompletedAndWorkRemaining( tracker.tryClaim(position); final Progress progress = tracker.getProgress(); - assertEquals(position.getSeconds(), progress.getWorkCompleted(), DELTA); - assertEquals(to.getSeconds() - position.getSeconds(), progress.getWorkRemaining(), DELTA); + assertEquals((double) position.getSeconds(), progress.getWorkCompleted(), DELTA); + assertEquals( + (double) (to.getSeconds() - position.getSeconds()), progress.getWorkRemaining(), DELTA); } @Test @@ -249,7 +250,7 @@ public void testGetProgressReturnsWorkRemainingAsWholeRangeWhenNoClaimWasAttempt final Progress progress = tracker.getProgress(); assertEquals(0D, progress.getWorkCompleted(), DELTA); - assertEquals(to.getSeconds(), progress.getWorkRemaining(), DELTA); + assertEquals((double) to.getSeconds(), progress.getWorkRemaining(), DELTA); } @Test diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/util/TestJsonMapper.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/util/TestJsonMapper.java index cd55cd4c2205..fefc1b639d1b 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/util/TestJsonMapper.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/util/TestJsonMapper.java @@ -111,12 +111,12 @@ private static Value recordValueFrom( .putFields( "number_of_records_in_transaction", Value.newBuilder() - .setNumberValue(record.getNumberOfRecordsInTransaction()) + .setNumberValue((double) record.getNumberOfRecordsInTransaction()) .build()) .putFields( "number_of_partitions_in_transaction", Value.newBuilder() - .setNumberValue(record.getNumberOfPartitionsInTransaction()) + .setNumberValue((double) record.getNumberOfPartitionsInTransaction()) .build()) .putFields( "transaction_tag", @@ -149,7 +149,9 @@ private static Value columnTypeValueFrom(ColumnType columnType) Value.newBuilder().setBoolValue(columnType.isPrimaryKey()).build()) .putFields( "ordinal_position", - Value.newBuilder().setNumberValue(columnType.getOrdinalPosition()).build()) + Value.newBuilder() + .setNumberValue((double) columnType.getOrdinalPosition()) + .build()) .build()) .build(); } diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java index d4810f88dfb6..6a214fc17533 100644 --- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java +++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaMetrics.java @@ -145,7 +145,7 @@ private void recordRpcLatencyMetrics() { // Update all the latencies for (Duration d : topicLatencies.getValue()) { Preconditions.checkArgumentNotNull(topicHistogram); - topicHistogram.update(d.toMillis()); + topicHistogram.update((double) d.toMillis()); } } } diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java index a05abba06e75..68ab866277fa 100644 --- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java +++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java @@ -666,7 +666,7 @@ public ProcessContinuation processElement( } } finally { if (rawSizesCount > 0L) { - avgRecordSize.update(rawSizesSum, rawSizesCount); + avgRecordSize.update((double) rawSizesSum, rawSizesCount); rawSizes.update(rawSizesSum, rawSizesCount, rawSizesMin, rawSizesMax); } } diff --git a/sdks/java/testing/load-tests/src/main/java/org/apache/beam/sdk/loadtests/LoadTest.java b/sdks/java/testing/load-tests/src/main/java/org/apache/beam/sdk/loadtests/LoadTest.java index aaa2691b7d0f..ad09fe27747c 100644 --- a/sdks/java/testing/load-tests/src/main/java/org/apache/beam/sdk/loadtests/LoadTest.java +++ b/sdks/java/testing/load-tests/src/main/java/org/apache/beam/sdk/loadtests/LoadTest.java @@ -174,7 +174,7 @@ private List readMetrics( testId, timestamp.toString(), buildMetric("total_bytes_count"), - reader.getCounterMetric("totalBytes.count")); + (double) reader.getCounterMetric("totalBytes.count")); return Arrays.asList(runtime, totalBytes); }