docs: public void initMultiCountMetrics(PhysicalPlanHelper helper) {
      // For bolt, we would consider both input stream and output stream
      List<TopologyAPI.InputStream> inputs = helper.getMyBolt().getInputsList();
      for (TopologyAPI.InputStream inputStream : inputs) {
        String streamId = inputStream.getStream().getId();
        String globalStreamId =
            new StringBuilder(inputStream.getStream().getComponentName()).
                append("/").append(streamId).toString();
  
        ackCount.scope(streamId);
        failCount.scope(streamId);
        executeCount.scope(streamId);
        executeTimeNs.scope(streamId);
  
        ackCount.scope(globalStreamId);
        failCount.scope(globalStreamId);
        executeCount.scope(globalStreamId);
        executeTimeNs.scope(globalStreamId);
      }
      List<TopologyAPI.OutputStream> outputs = helper.getMyBolt().getOutputsList();
      for (TopologyAPI.OutputStream outputStream : outputs) {
        String streamId = outputStream.getStream().getId();
        emitCount.scope(streamId);
      }
    }