This is an automated email from the ASF dual-hosted git repository.
raboof pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new a7ca62d4 add hashCode impl on KafkaAsyncConsumerCommitterRef (#625)
a7ca62d4 is described below
commit a7ca62d4354679ad091c56773c91f2ec54f6a235
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 10 08:37:47 2026 +0100
add hashCode impl on KafkaAsyncConsumerCommitterRef (#625)
* add hashCode impl on KafkaAsyncConsumerCommitterRef
* scalafmt
---
.../pekko/kafka/internal/CommittableSources.scala | 3 +
.../KafkaAsyncConsumerCommitterRefSpec.scala | 88 ++++++++++++++++++++++
2 files changed, 91 insertions(+)
diff --git
a/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
b/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
index f5934482..ee08377c 100644
---
a/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
+++
b/core/src/main/scala/org/apache/pekko/kafka/internal/CommittableSources.scala
@@ -228,6 +228,9 @@ private[kafka] class KafkaAsyncConsumerCommitterRef(private
val consumerActor: A
this.consumerActor == that.consumerActor && this.commitTimeout ==
that.commitTimeout
case _ => false
}
+
+ override def hashCode(): Int =
+ java.util.Objects.hash(consumerActor, commitTimeout)
}
@InternalApi
diff --git
a/tests/src/test/scala/org/apache/pekko/kafka/internal/KafkaAsyncConsumerCommitterRefSpec.scala
b/tests/src/test/scala/org/apache/pekko/kafka/internal/KafkaAsyncConsumerCommitterRefSpec.scala
new file mode 100644
index 00000000..f8418d37
--- /dev/null
+++
b/tests/src/test/scala/org/apache/pekko/kafka/internal/KafkaAsyncConsumerCommitterRefSpec.scala
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.kafka.internal
+
+import org.apache.pekko
+import pekko.actor.{ ActorRef, ActorSystem, Props }
+import pekko.testkit.TestKit
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+import scala.concurrent.duration._
+
+class KafkaAsyncConsumerCommitterRefSpec
+ extends TestKit(ActorSystem("KafkaAsyncConsumerCommitterRefSpec"))
+ with AnyWordSpecLike
+ with Matchers
+ with BeforeAndAfterAll {
+
+ override def afterAll(): Unit =
+ TestKit.shutdownActorSystem(system)
+
+ private def committer(actorRef: ActorRef, timeout: FiniteDuration) =
+ new KafkaAsyncConsumerCommitterRef(actorRef, timeout)(system.dispatcher)
+
+ "KafkaAsyncConsumerCommitterRef equals/hashCode" should {
+
+ "be equal for same actorRef and timeout" in {
+ val ref = system.actorOf(Props.empty)
+ val a = committer(ref, 5.seconds)
+ val b = committer(ref, 5.seconds)
+ a shouldEqual b
+ a.hashCode() shouldEqual b.hashCode()
+ }
+
+ "not be equal for different actorRef" in {
+ val ref1 = system.actorOf(Props.empty)
+ val ref2 = system.actorOf(Props.empty)
+ val a = committer(ref1, 5.seconds)
+ val b = committer(ref2, 5.seconds)
+ (a should not).equal(b)
+ }
+
+ "not be equal for different timeout" in {
+ val ref = system.actorOf(Props.empty)
+ val a = committer(ref, 5.seconds)
+ val b = committer(ref, 10.seconds)
+ (a should not).equal(b)
+ }
+
+ "not be equal to a non-KafkaAsyncConsumerCommitterRef" in {
+ val ref = system.actorOf(Props.empty)
+ val a = committer(ref, 5.seconds)
+ (a should not).equal("not a committer")
+ }
+
+ "have consistent hashCode for equal instances" in {
+ val ref = system.actorOf(Props.empty)
+ val a = committer(ref, 150.millis)
+ val b = committer(ref, 150.millis)
+ a.hashCode() shouldEqual b.hashCode()
+ }
+
+ "have different hashCode for different actorRef (probabilistic)" in {
+ val ref1 = system.actorOf(Props.empty)
+ val ref2 = system.actorOf(Props.empty)
+ val a = committer(ref1, 5.seconds)
+ val b = committer(ref2, 5.seconds)
+ // Not guaranteed but extremely likely
+ (a.hashCode() should not).equal(b.hashCode())
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
