You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2024/01/03 08:18:29 UTC
(incubator-celeborn) branch branch-0.4 updated: [CELEBORN-1190][FOLLOWUP] Apply error prone patch and suppress some problems
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.4 by this push:
new e630a55f5 [CELEBORN-1190][FOLLOWUP] Apply error prone patch and suppress some problems
e630a55f5 is described below
commit e630a55f59ce8903bfb08966dcf3b9a07aba54e1
Author: sychen <sy...@ctrip.com>
AuthorDate: Wed Jan 3 15:59:50 2024 +0800
[CELEBORN-1190][FOLLOWUP] Apply error prone patch and suppress some problems
1. Fix IntLongMath, InconsistentCapitalization, UnnecessaryAssignment
2. disable StringSplitter, EmptyBlockTag, EqualsGetClass, MissingSummary, BadImport
No
GA
```bash
./build/make-distribution.sh --release
```
PR
total 202
```
43 SynchronizeOnNonFinalField
24 StaticAssignmentInConstructor
21 JdkObsolete
16 ThreadLocalUsage
16 MutableConstantField
16 MissingCasesInEnumSwitch
11 UnusedMethod
11 NonAtomicVolatileUpdate
8 UnusedNestedClass
8 NonOverridingEquals
8 Finally
4 MixedMutabilityReturnType
4 DoubleBraceInitialization
4 CatchAndPrintStackTrace
4 CanonicalDuration
2 ReferenceEquality
1 ClassCanBeStatic
1 ByteBufferBackingArray
```
Closes #2180 from cxzl25/error_prone_patch_followup.
Authored-by: sychen <sy...@ctrip.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../plugin/flink/FlinkResultPartitionInfo.java | 4 ++--
.../celeborn/plugin/flink/buffer/BufferPacker.java | 1 +
.../plugin/flink/buffer/PartitionSortedBuffer.java | 2 +-
.../plugin/flink/RemoteShuffleEnvironment.java | 4 ++--
.../plugin/flink/RemoteShuffleResultPartition.java | 3 ++-
.../plugin/flink/RemoteShuffleEnvironment.java | 4 ++--
.../plugin/flink/RemoteShuffleResultPartition.java | 3 ++-
.../plugin/flink/RemoteShuffleEnvironment.java | 4 ++--
.../plugin/flink/RemoteShuffleResultPartition.java | 3 ++-
.../plugin/flink/RemoteShuffleEnvironment.java | 4 ++--
.../plugin/flink/RemoteShuffleResultPartition.java | 3 ++-
.../shuffle/celeborn/SortBasedShuffleWriter.java | 2 ++
.../celeborn/CelebornShuffleWriterSuiteBase.java | 8 ++++----
.../shuffle/celeborn/HashBasedShuffleWriter.java | 1 +
.../shuffle/celeborn/SortBasedShuffleWriter.java | 1 +
.../shuffle/celeborn/SparkShuffleManager.java | 2 ++
.../celeborn/CelebornShuffleWriterSuiteBase.java | 8 ++++----
.../common/network/util/TransportFrameDecoder.java | 4 ++--
.../deploy/master/clustermeta/ha/MetaHandler.java | 1 -
.../deploy/master/SlotsAllocatorSuiteJ.java | 2 +-
pom.xml | 22 ++++++++++++++++++++--
.../deploy/worker/storage/CreditStreamManager.java | 6 +++---
.../worker/storage/PartitionFilesSorter.java | 2 +-
23 files changed, 61 insertions(+), 33 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
index 1d8d81857..a8591d3e2 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/FlinkResultPartitionInfo.java
@@ -30,8 +30,8 @@ public class FlinkResultPartitionInfo {
private final ProducerDescriptor producerDescriptor;
public FlinkResultPartitionInfo(
- JobID jobId, PartitionDescriptor partitionDescriptor, ProducerDescriptor producerDescriptor) {
- this.jobID = jobId;
+ JobID jobID, PartitionDescriptor partitionDescriptor, ProducerDescriptor producerDescriptor) {
+ this.jobID = jobID;
this.partitionDescriptor = partitionDescriptor;
this.producerDescriptor = producerDescriptor;
}
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/BufferPacker.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/BufferPacker.java
index c0176e111..d0a757f19 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/BufferPacker.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/BufferPacker.java
@@ -219,6 +219,7 @@ public class BufferPacker {
// Flink 1.18.0
// [FLINK-32549][network] Tiered storage memory manager supports ownership transfer for buffers
+ @SuppressWarnings("MissingOverride")
public void setRecycler(BufferRecycler bufferRecycler) {
DynMethods.builder("setRecycler")
.impl(buffer.getClass().getName(), BufferRecycler.class)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
index 88b5338d6..c11e5d373 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java
@@ -325,7 +325,7 @@ public class PartitionSortedBuffer implements SortBuffer {
// skip the data already read if there is remaining partial record after the previous
// copy
long position = (long) sourceSegmentOffset + (recordLength - recordRemainingBytes);
- sourceSegmentIndex += (position / bufferSize);
+ sourceSegmentIndex = (int) (sourceSegmentIndex + (position / bufferSize));
sourceSegmentOffset = (int) (position % bufferSize);
} else {
recordRemainingBytes = recordLength;
diff --git a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
index 2760f88f0..32f8873a3 100644
--- a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
+++ b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
@@ -44,8 +44,8 @@ public class RemoteShuffleEnvironment extends AbstractRemoteShuffleEnvironment
/**
* @param networkBufferPool Network buffer pool for shuffle read and shuffle write.
* @param resultPartitionManager A trivial {@link ResultPartitionManager}.
- * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}. //
- * * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
+ * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}.
+ * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
*/
public RemoteShuffleEnvironment(
NetworkBufferPool networkBufferPool,
diff --git a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
index bae30ca4f..d28e78285 100644
--- a/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
+++ b/client-flink/flink-1.14/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
@@ -202,7 +202,8 @@ public class RemoteShuffleResultPartition extends ResultPartition {
public void updateStatistics(
SortBuffer.BufferWithChannel bufferWithChannel, boolean isBroadcast) {
numBuffersOut.inc(isBroadcast ? numSubpartitions : 1);
- long readableBytes = bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
+ long readableBytes =
+ (long) bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
numBytesOut.inc(isBroadcast ? readableBytes * numSubpartitions : readableBytes);
}
}
diff --git a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
index 01740f1f3..f76a18478 100644
--- a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
+++ b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
@@ -44,8 +44,8 @@ public class RemoteShuffleEnvironment extends AbstractRemoteShuffleEnvironment
/**
* @param networkBufferPool Network buffer pool for shuffle read and shuffle write.
* @param resultPartitionManager A trivial {@link ResultPartitionManager}.
- * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}. //
- * * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
+ * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}.
+ * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
*/
public RemoteShuffleEnvironment(
NetworkBufferPool networkBufferPool,
diff --git a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
index 63106a229..3b25a4a0a 100644
--- a/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
+++ b/client-flink/flink-1.15/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
@@ -202,7 +202,8 @@ public class RemoteShuffleResultPartition extends ResultPartition {
public void updateStatistics(
SortBuffer.BufferWithChannel bufferWithChannel, boolean isBroadcast) {
numBuffersOut.inc(isBroadcast ? numSubpartitions : 1);
- long readableBytes = bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
+ long readableBytes =
+ (long) bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
numBytesProduced.inc(readableBytes);
numBytesOut.inc(isBroadcast ? readableBytes * numSubpartitions : readableBytes);
}
diff --git a/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java b/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
index 01740f1f3..f76a18478 100644
--- a/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
+++ b/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
@@ -44,8 +44,8 @@ public class RemoteShuffleEnvironment extends AbstractRemoteShuffleEnvironment
/**
* @param networkBufferPool Network buffer pool for shuffle read and shuffle write.
* @param resultPartitionManager A trivial {@link ResultPartitionManager}.
- * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}. //
- * * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
+ * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}.
+ * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
*/
public RemoteShuffleEnvironment(
NetworkBufferPool networkBufferPool,
diff --git a/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java b/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
index 820e456db..c888b6ee9 100644
--- a/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
+++ b/client-flink/flink-1.17/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
@@ -214,7 +214,8 @@ public class RemoteShuffleResultPartition extends ResultPartition {
public void updateStatistics(
SortBuffer.BufferWithChannel bufferWithChannel, boolean isBroadcast) {
numBuffersOut.inc(isBroadcast ? numSubpartitions : 1);
- long readableBytes = bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
+ long readableBytes =
+ (long) bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
if (isBroadcast) {
resultPartitionBytes.incAll(readableBytes);
} else {
diff --git a/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java b/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
index 01740f1f3..f76a18478 100644
--- a/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
+++ b/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleEnvironment.java
@@ -44,8 +44,8 @@ public class RemoteShuffleEnvironment extends AbstractRemoteShuffleEnvironment
/**
* @param networkBufferPool Network buffer pool for shuffle read and shuffle write.
* @param resultPartitionManager A trivial {@link ResultPartitionManager}.
- * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}. //
- * * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
+ * @param resultPartitionFactory Factory class to create {@link RemoteShuffleResultPartition}.
+ * @param inputGateFactory Factory class to create {@link RemoteShuffleInputGate}.
*/
public RemoteShuffleEnvironment(
NetworkBufferPool networkBufferPool,
diff --git a/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java b/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
index 820e456db..c888b6ee9 100644
--- a/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
+++ b/client-flink/flink-1.18/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleResultPartition.java
@@ -214,7 +214,8 @@ public class RemoteShuffleResultPartition extends ResultPartition {
public void updateStatistics(
SortBuffer.BufferWithChannel bufferWithChannel, boolean isBroadcast) {
numBuffersOut.inc(isBroadcast ? numSubpartitions : 1);
- long readableBytes = bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
+ long readableBytes =
+ (long) bufferWithChannel.getBuffer().readableBytes() - BufferUtils.HEADER_LENGTH;
if (isBroadcast) {
resultPartitionBytes.incAll(readableBytes);
} else {
diff --git a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
index daf251047..988cdc090 100644
--- a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
+++ b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
@@ -405,6 +405,8 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
}
}
+ // Added in SPARK-32917, for Spark 3.2 and above
+ @SuppressWarnings("MissingOverride")
public long[] getPartitionLengths() {
throw new UnsupportedOperationException(
"Celeborn is not compatible with Spark push mode, please set spark.shuffle.push.enabled to false");
diff --git a/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java b/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java
index f6f38a592..cc0224e79 100644
--- a/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java
+++ b/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java
@@ -106,16 +106,16 @@ public abstract class CelebornShuffleWriterSuiteBase {
SparkUtils.createMapStatus(bmId, new long[numPartitions], new long[numPartitions]);
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- private TaskContext taskContext = null;
+ private TaskContext taskContext;
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- private ShuffleDependency<Integer, String, String> dependency = null;
+ private ShuffleDependency<Integer, String, String> dependency;
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- private SparkEnv env = null;
+ private SparkEnv env;
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- private BlockManager blockManager = null;
+ private BlockManager blockManager;
private TaskMetrics metrics = null;
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
index a7c494c03..a3024bf00 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
@@ -408,6 +408,7 @@ public class HashBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
}
// Added in SPARK-32917, for Spark 3.2 and above
+ @SuppressWarnings("MissingOverride")
public long[] getPartitionLengths() {
throw new UnsupportedOperationException(
"Celeborn is not compatible with Spark push mode, please set spark.shuffle.push.enabled to false");
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
index a2f3cdfb7..2de03a569 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
@@ -427,6 +427,7 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
}
// Added in SPARK-32917, for Spark 3.2 and above
+ @SuppressWarnings("MissingOverride")
public long[] getPartitionLengths() {
throw new UnsupportedOperationException(
"Celeborn is not compatible with push-based shuffle, please set spark.shuffle.push.enabled to false");
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
index 03a38b933..e130566f6 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
@@ -318,6 +318,7 @@ public class SparkShuffleManager implements ShuffleManager {
}
// Added in SPARK-32055, for Spark 3.1 and above
+ @SuppressWarnings("MissingOverride")
public <K, C> ShuffleReader<K, C> getReader(
ShuffleHandle handle,
int startMapIndex,
@@ -342,6 +343,7 @@ public class SparkShuffleManager implements ShuffleManager {
}
// Marked as final in SPARK-32055, reserved for Spark 3.0
+ @SuppressWarnings("MissingOverride")
public <K, C> ShuffleReader<K, C> getReader(
ShuffleHandle handle,
int startPartition,
diff --git a/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java b/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java
index 65930d9ba..adf3cda84 100644
--- a/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java
+++ b/client-spark/spark-3/src/test/java/org/apache/spark/shuffle/celeborn/CelebornShuffleWriterSuiteBase.java
@@ -107,16 +107,16 @@ public abstract class CelebornShuffleWriterSuiteBase {
new TaskMemoryManager(UnifiedMemoryManager.apply(sparkConf, 1), 0);
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- protected TaskContext taskContext = null;
+ protected TaskContext taskContext;
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- protected ShuffleDependency<Integer, String, String> dependency = null;
+ protected ShuffleDependency<Integer, String, String> dependency;
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- private SparkEnv env = null;
+ private SparkEnv env;
@Mock(answer = Answers.RETURNS_SMART_NULLS)
- private BlockManager blockManager = null;
+ private BlockManager blockManager;
protected TaskMetrics metrics = null;
diff --git a/common/src/main/java/org/apache/celeborn/common/network/util/TransportFrameDecoder.java b/common/src/main/java/org/apache/celeborn/common/network/util/TransportFrameDecoder.java
index dec5db28b..83c89afbd 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/util/TransportFrameDecoder.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/util/TransportFrameDecoder.java
@@ -97,7 +97,7 @@ public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implemen
msgSize = first.readInt();
curType = Message.Type.decode(first);
bodySize = first.readInt();
- nextFrameSize = msgSize + bodySize;
+ nextFrameSize = (long) msgSize + bodySize;
totalSize -= HEADER_SIZE;
if (!first.isReadable()) {
buffers.removeFirst().release();
@@ -117,7 +117,7 @@ public class TransportFrameDecoder extends ChannelInboundHandlerAdapter implemen
msgSize = headerBuf.readInt();
curType = Message.Type.decode(headerBuf);
bodySize = headerBuf.readInt();
- nextFrameSize = msgSize + bodySize;
+ nextFrameSize = (long) msgSize + bodySize;
totalSize -= HEADER_SIZE;
return nextFrameSize;
}
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
index 13990bccd..ae660ed17 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
@@ -97,7 +97,6 @@ public class MetaHandler {
int replicatePort;
Map<String, DiskInfo> diskInfos;
Map<UserIdentifier, ResourceConsumption> userResourceConsumption;
- List<Map<String, Integer>> slots = new ArrayList<>();
Map<String, Long> estimatedAppDiskUsage = new HashMap<>();
switch (cmdType) {
case RequestSlots:
diff --git a/master/src/test/java/org/apache/celeborn/service/deploy/master/SlotsAllocatorSuiteJ.java b/master/src/test/java/org/apache/celeborn/service/deploy/master/SlotsAllocatorSuiteJ.java
index 9ca86fe77..511046922 100644
--- a/master/src/test/java/org/apache/celeborn/service/deploy/master/SlotsAllocatorSuiteJ.java
+++ b/master/src/test/java/org/apache/celeborn/service/deploy/master/SlotsAllocatorSuiteJ.java
@@ -272,7 +272,7 @@ public class SlotsAllocatorSuiteJ {
}
int allocateToDiskSlots = 0;
for (WorkerInfo worker : workers) {
- allocateToDiskSlots += worker.usedSlots();
+ allocateToDiskSlots = (int) (allocateToDiskSlots + worker.usedSlots());
}
if (shouldReplicate) {
assertTrue(partitionIds.size() * 2 >= unknownDiskSlots + allocateToDiskSlots);
diff --git a/pom.xml b/pom.xml
index f120f6889..a836966a3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -929,7 +929,16 @@
<fork>true</fork>
<compilerArgs>
<arg>-XDcompilePolicy=simple</arg>
- <arg>-Xplugin:ErrorProne -XepExcludedPaths:.*/target/generated-sources/protobuf/.* -Xep:FutureReturnValueIgnored:OFF -Xep:TypeParameterUnusedInFormals:OFF -Xep:UnusedVariable:OFF</arg>
+ <arg>-Xplugin:ErrorProne \
+ -XepExcludedPaths:.*/target/generated-sources/protobuf/.* \
+ -Xep:FutureReturnValueIgnored:OFF \
+ -Xep:TypeParameterUnusedInFormals:OFF \
+ -Xep:UnusedVariable:OFF \
+ -Xep:StringSplitter:OFF \
+ -Xep:EmptyBlockTag:OFF \
+ -Xep:EqualsGetClass:OFF \
+ -Xep:MissingSummary:OFF \
+ -Xep:BadImport:OFF</arg>
</compilerArgs>
<annotationProcessorPaths>
<path>
@@ -1238,7 +1247,16 @@
<encoding>UTF-8</encoding>
<compilerArgs>
<arg>-XDcompilePolicy=simple</arg>
- <arg>-Xplugin:ErrorProne</arg>
+ <arg>-Xplugin:ErrorProne \
+ -XepExcludedPaths:.*/target/generated-sources/protobuf/.* \
+ -Xep:FutureReturnValueIgnored:OFF \
+ -Xep:TypeParameterUnusedInFormals:OFF \
+ -Xep:UnusedVariable:OFF \
+ -Xep:StringSplitter:OFF \
+ -Xep:EmptyBlockTag:OFF \
+ -Xep:EqualsGetClass:OFF \
+ -Xep:MissingSummary:OFF \
+ -Xep:BadImport:OFF</arg>
<arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED</arg>
<arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED</arg>
<arg>-J--add-exports=jdk.compiler/com.sun.tools.javac.main=ALL-UNNAMED</arg>
diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManager.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManager.java
index bf93bbeb8..24c59e86c 100644
--- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManager.java
+++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/CreditStreamManager.java
@@ -55,21 +55,21 @@ public class CreditStreamManager {
public CreditStreamManager(
int minReadBuffers,
int maxReadBuffers,
- int threadsPerMountpoint,
+ int threadsPerMountPoint,
int minBuffersToTriggerRead) {
nextStreamId = new AtomicLong((long) new Random().nextInt(Integer.MAX_VALUE) * 1000);
streams = JavaUtils.newConcurrentHashMap();
activeMapPartitions = JavaUtils.newConcurrentHashMap();
this.minReadBuffers = minReadBuffers;
this.maxReadBuffers = maxReadBuffers;
- threadsPerMountPoint = threadsPerMountpoint;
+ this.threadsPerMountPoint = threadsPerMountPoint;
this.minBuffersToTriggerRead = minBuffersToTriggerRead;
MemoryManager.instance().setCreditStreamManager(this);
logger.debug(
"Initialize buffer stream manager with {} {} {}",
this.minReadBuffers,
this.maxReadBuffers,
- threadsPerMountpoint);
+ threadsPerMountPoint);
}
public long registerStream(
diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
index b229b89e3..d072d259a 100644
--- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
+++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
@@ -579,7 +579,7 @@ public class PartitionFilesSorter extends ShuffleRecoverHelper {
originShuffleBlockInfos.computeIfAbsent(mapId, v -> new ArrayList<>());
ShuffleBlockInfo blockInfo = new ShuffleBlockInfo();
blockInfo.offset = blockStartIndex;
- blockInfo.length = compressedSize + 16;
+ blockInfo.length = compressedSize + 16L;
singleMapIdShuffleBlockList.add(blockInfo);
index += batchHeaderLen + compressedSize;