Simplify: deduplicate sendControlMessage, HashMap over ConcurrentHashMap,

rename test

Code review findings:
- VeniceWriter: make 5-param sendControlMessage delegate to 6-param overload,
  eliminating duplicated getDebugInfo/isEndOfSegment/synchronized logic
- MapLongAccumulator: ConcurrentHashMap → HashMap — Spark tasks are
  single-threaded (each gets a fresh copy()), no concurrent access occurs
- MRJobCounterHelperTest: rename misleading test to reflect what it actually
  tests (counter retrieval round-trip, not Reporter-based increment)

Author: sushantmane

clients/venice-push-job/src/main/java/com/linkedin/venice/spark/datawriter/task/MapLongAccumulator.java

@@ -1,20 +1,23 @@
 package com.linkedin.venice.spark.datawriter.task;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import org.apache.spark.SparkContext;
 import org.apache.spark.util.AccumulatorV2;
 
 
 /**
  * A Spark accumulator that maintains per-key long counters, merging by summing values per key.
  * Used for tracking per-partition record counts during Spark-based push jobs.
+ *
+ * Thread safety: Spark creates a fresh copy per task via {@link #copy()}, so each task's
+ * instance is single-threaded. A plain HashMap suffices — no concurrent access occurs.
  */
 public class MapLongAccumulator extends AccumulatorV2<scala.Tuple2<Integer, Long>, Map<Integer, Long>> {
   private static final long serialVersionUID = 1L;
 
-  private final ConcurrentHashMap<Integer, Long> map = new ConcurrentHashMap<>();
+  private final HashMap<Integer, Long> map = new HashMap<>();
 
   MapLongAccumulator() {
   }

clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/mapreduce/counter/MRJobCounterHelperTest.java

@@ -33,9 +33,7 @@ public void testGetPerPartitionRecordCountsEmpty() {
   }
 
   @Test
-  public void testIncrPartitionRecordCountWithMockReporter() {
-    // Use a Counters-based approach to verify the increment path indirectly:
-    // Increment via Reporter, then read back via Counters
+  public void testGetPerPartitionRecordCountsWithMultipleIncrements() {
     Counters counters = new Counters();
     Counters.Counter counter = counters.findCounter("Per Partition Record Count", "3");
     counter.increment(10);

internal/venice-common/src/main/java/com/linkedin/venice/writer/VeniceWriter.java

@@ -2277,23 +2277,7 @@ public CompletableFuture<PubSubProduceResult> sendControlMessage(
       Map<String, String> debugInfo,
       PubSubProducerCallback callback,
       LeaderMetadataWrapper leaderMetadataWrapper) {
-    // Work around until we upgrade to a more modern Avro version which supports overriding the
-    // String implementation.
-    controlMessage.debugInfo = getDebugInfo(debugInfo);
-    boolean isEndOfSegment = ControlMessageType.valueOf(controlMessage).equals(ControlMessageType.END_OF_SEGMENT);
-    synchronized (this.partitionLocks[partition]) {
-      return sendMessage(
-          this::getControlMessageKey,
-          MessageType.CONTROL_MESSAGE,
-          controlMessage,
-          isEndOfSegment,
-          partition,
-          callback,
-          true,
-          leaderMetadataWrapper,
-          VENICE_DEFAULT_LOGICAL_TS,
-          EmptyPubSubMessageHeaders.SINGLETON);
-    }
+    return sendControlMessage(controlMessage, partition, debugInfo, callback, leaderMetadataWrapper, null);
   }
 
   public CompletableFuture<PubSubProduceResult> sendControlMessage(
@@ -2303,6 +2287,8 @@ public CompletableFuture<PubSubProduceResult> sendControlMessage(
       PubSubProducerCallback callback,
       LeaderMetadataWrapper leaderMetadataWrapper,
       PubSubMessageHeaders headers) {
+    // Work around until we upgrade to a more modern Avro version which supports overriding the
+    // String implementation.
     controlMessage.debugInfo = getDebugInfo(debugInfo);
     boolean isEndOfSegment = ControlMessageType.valueOf(controlMessage).equals(ControlMessageType.END_OF_SEGMENT);
     synchronized (this.partitionLocks[partition]) {