[Dataflow] Remove unused PhysicalElementCount Counter (#34461)

Author: arunpandianp

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java

@@ -28,15 +28,12 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 
 /**
- * A Dataflow-specific version of {@link ElementCounter}, which specifies the object counter name
- * differently as PhysicalElementCount. Additionally, it counts element windows as ElementCount.
+ * A Dataflow-specific version of {@link ElementCounter}. It counts element windows as ElementCount.
  */
 @SuppressWarnings({
   "nullness" // TODO(https://github.com/apache/beam/issues/20497)
 })
 public class DataflowOutputCounter implements ElementCounter {
-  /** Number of physical element and multiple-window assignments that were serialized/processed. */
-  private static final String OBJECT_COUNTER_NAME = "-PhysicalElementCount";
   /** Number of logical element and single window pairs that were processed. */
   private static final String ELEMENT_COUNTER_NAME = "-ElementCount";
 
@@ -57,7 +54,6 @@ public DataflowOutputCounter(
       NameContext nameContext) {
     objectAndByteCounter =
         new OutputObjectAndByteCounter(elementByteSizeObservable, counterFactory, nameContext);
-    objectAndByteCounter.countObject(outputName + OBJECT_COUNTER_NAME);
     objectAndByteCounter.countMeanByte(outputName + MEAN_BYTE_COUNTER_NAME);
     createElementCounter(counterFactory, outputName + ELEMENT_COUNTER_NAME);
   }
@@ -86,11 +82,6 @@ static String getElementCounterName(String prefix) {
     return prefix + ELEMENT_COUNTER_NAME;
   }
 
-  @VisibleForTesting
-  static String getObjectCounterName(String prefix) {
-    return prefix + OBJECT_COUNTER_NAME;
-  }
-
   @VisibleForTesting
   static String getMeanByteCounterName(String prefix) {
     return prefix + MEAN_BYTE_COUNTER_NAME;

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactoryTest.java

@@ -20,7 +20,6 @@
 import static org.apache.beam.runners.dataflow.util.Structs.addString;
 import static org.apache.beam.runners.dataflow.worker.DataflowOutputCounter.getElementCounterName;
 import static org.apache.beam.runners.dataflow.worker.DataflowOutputCounter.getMeanByteCounterName;
-import static org.apache.beam.runners.dataflow.worker.DataflowOutputCounter.getObjectCounterName;
 import static org.apache.beam.runners.dataflow.worker.counters.CounterName.named;
 import static org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray;
 import static org.apache.beam.sdk.util.StringUtils.byteArrayToJsonString;
@@ -242,8 +241,6 @@ private static void verifyOutputCounters(
     for (String outputName : outputNames) {
       verify(updateExtractor)
           .longSum(eq(named(getElementCounterName(outputName))), anyBoolean(), anyLong());
-      verify(updateExtractor)
-          .longSum(eq(named(getObjectCounterName(outputName))), anyBoolean(), anyLong());
       verify(updateExtractor)
           .longMean(
               eq(named(getMeanByteCounterName(outputName))),