Merge pull request #25035 from piotrromanski/wip-fix-math-abs-usage

Handle a negative value returned by Math.abs()

akka-actor/src/main/scala/akka/dispatch/affinity/AffinityPool.scala

@@ -401,7 +401,7 @@ private[akka] final class FairDistributionHashCache( final val config: Config) e
 
   override final def create(): QueueSelector = new AtomicReference[ImmutableIntMap](ImmutableIntMap.empty) with QueueSelector {
     override def toString: String = s"FairDistributionHashCache(fairDistributionThreshold = $fairDistributionThreshold)"
-    private[this] final def improve(h: Int): Int = Math.abs(reverseBytes(h * 0x9e3775cd) * 0x9e3775cd) // `sbhash`: In memory of Phil Bagwell.
+    private[this] final def improve(h: Int): Int = 0x7FFFFFFF & (reverseBytes(h * 0x9e3775cd) * 0x9e3775cd) // `sbhash`: In memory of Phil Bagwell.
     override final def getQueue(command: Runnable, queues: Int): Int = {
       val runnableHash = command.hashCode()
       if (fairDistributionThreshold == 0)

akka-cluster-sharding/src/main/scala/akka/cluster/sharding/Shard.scala

@@ -607,7 +607,7 @@ private[akka] class DDataShard(
     Array.tabulate(numberOfKeys)(i ⇒ ORSetKey[EntityId](s"shard-${typeName}-${shardId}-$i"))
 
   private def key(entityId: EntityId): ORSetKey[EntityId] = {
-    val i = (math.abs(entityId.hashCode) % numberOfKeys)
+    val i = (math.abs(entityId.hashCode % numberOfKeys))
     stateKeys(i)
   }
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala

@@ -1657,7 +1657,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
         }
         val chunk = (statusCount % totChunks).toInt
         val status = Status(dataEntries.collect {
-          case (key, (_, _)) if math.abs(key.hashCode) % totChunks == chunk ⇒ (key, getDigest(key))
+          case (key, (_, _)) if math.abs(key.hashCode % totChunks) == chunk ⇒ (key, getDigest(key))
         }, chunk, totChunks)
         to ! status
       }
@@ -1685,7 +1685,7 @@ final class Replicator(settings: ReplicatorSettings) extends Actor with ActorLog
     val otherKeys = otherDigests.keySet
     val myKeys =
       if (totChunks == 1) dataEntries.keySet
-      else dataEntries.keysIterator.filter(key ⇒ math.abs(key.hashCode) % totChunks == chunk).toSet
+      else dataEntries.keysIterator.filter(key ⇒ math.abs(key.hashCode % totChunks) == chunk).toSet
     val otherMissingKeys = myKeys diff otherKeys
     val keys = (otherDifferentKeys ++ otherMissingKeys).take(maxDeltaElements)
     if (keys.nonEmpty) {

akka-docs/src/test/java/jdocs/stream/HubDocTest.java

@@ -158,7 +158,7 @@ public void dynamicPartition() {
     RunnableGraph<Source<String, NotUsed>> runnableGraph =
       producer.toMat(PartitionHub.of(
           String.class, 
-          (size, elem) -> Math.abs(elem.hashCode()) % size,
+          (size, elem) -> Math.abs(elem.hashCode() % size),
           2, 256), Keep.right());
 
     // By running/materializing the producer, we get back a Source, which

akka-docs/src/test/scala/docs/stream/HubsDocSpec.scala

@@ -118,7 +118,7 @@ class HubsDocSpec extends AkkaSpec with CompileOnlySpec {
       // value to the left is used)
       val runnableGraph: RunnableGraph[Source[String, NotUsed]] =
         producer.toMat(PartitionHub.sink(
-          (size, elem) ⇒ math.abs(elem.hashCode) % size,
+          (size, elem) ⇒ math.abs(elem.hashCode % size),
           startAfterNrOfConsumers = 2, bufferSize = 256))(Keep.right)
 
       // By running/materializing the producer, we get back a Source, which

akka-remote/src/main/scala/akka/remote/artery/ArteryTransport.scala

@@ -486,7 +486,7 @@ private[remote] abstract class ArteryTransport(_system: ExtendedActorSystem, _pr
         val b = env.originUid
         val hashA = 23 + a
         val hash: Int = 23 * hashA + java.lang.Long.hashCode(b)
-        math.abs(hash) % inboundLanes
+        math.abs(hash % inboundLanes)
       case OptionVal.None ⇒
         // the lane is set by the DuplicateHandshakeReq stage, otherwise 0
         env.lane

akka-remote/src/main/scala/akka/remote/artery/Association.scala

@@ -409,7 +409,7 @@ private[remote] class Association(
                 OrdinaryQueueIndex
               } else {
                 // select lane based on destination, to preserve message order
-                OrdinaryQueueIndex + (math.abs(r.path.uid) % outboundLanes)
+                OrdinaryQueueIndex + (math.abs(r.path.uid % outboundLanes))
               }
             r.cachedSendQueueIndex = idx
             idx

akka-stream/src/main/scala/akka/stream/javadsl/Hub.scala

@@ -172,7 +172,7 @@ object PartitionHub {
    * @param partitioner Function that decides where to route an element. The function takes two parameters;
    *   the first is the number of active consumers and the second is the stream element. The function should
    *   return the index of the selected consumer for the given element, i.e. int greater than or equal to 0
-   *   and less than number of consumers. E.g. `(size, elem) -> Math.abs(elem.hashCode()) % size`. It's also
+   *   and less than number of consumers. E.g. `(size, elem) -> Math.abs(elem.hashCode() % size)`. It's also
    *   possible to use `-1` to drop the element.
    * @param startAfterNrOfConsumers Elements are buffered until this number of consumers have been connected.
    *   This is only used initially when the operator is starting up, i.e. it is not honored when consumers have

akka-stream/src/main/scala/akka/stream/scaladsl/Hub.scala

@@ -789,7 +789,7 @@ object PartitionHub {
    * @param partitioner Function that decides where to route an element. The function takes two parameters;
    *   the first is the number of active consumers and the second is the stream element. The function should
    *   return the index of the selected consumer for the given element, i.e. int greater than or equal to 0
-   *   and less than number of consumers. E.g. `(size, elem) => math.abs(elem.hashCode) % size`. It's also
+   *   and less than number of consumers. E.g. `(size, elem) => math.abs(elem.hashCode % size)`. It's also
    *   possible to use `-1` to drop the element.
    * @param startAfterNrOfConsumers Elements are buffered until this number of consumers have been connected.
    *   This is only used initially when the operator is starting up, i.e. it is not honored when consumers have