code: @Override
      public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, TridentCollector collector)
      {
          long now = getCurrentTime();
          if (now - this.lastRotate > this.rotateTime)
          {
              Map<TransactionAttempt, DrpcRequestInfo> failed = this.idsMap.rotate();
              for (Entry<TransactionAttempt, DrpcRequestInfo> entry : failed.entrySet())
              {
                  fail(entry.getKey(), entry.getValue());
              }
              this.lastRotate = now;
          }
  
          if (this.idsMap.containsKey(tx) == true)
          {
              DrpcRequestInfo requestInfo = (DrpcRequestInfo) this.idsMap.remove(tx);
              fail(tx, requestInfo);
          }
  
          // 初期化未実施の場合初期化処理を行う。
          if (this.prepared == false)
          {
              prepare(this.stormConf, this.context);
              this.fetchHelper = createFetchHelper();
              this.fetchHelper.initialize(this.stormConf, this.function);
              this.prepared = true;
          }
  
          // DRPCリクエストを取得する。リクエストが存在しない場合、何も処理は行わない。
          DrpcRequestInfo requestInfo = null;
          try
          {
              requestInfo = this.fetchHelper.fetch();
          }
          catch (TException ex)
          {
              logger.warn("DRPC fetch failed.", ex);
          }
  
          if (requestInfo != null)
          {
              if (logger.isDebugEnabled() == true)
              {
                  String logFormat = "Request info get succeeded. TransactionAttempt={0}, DrpcRequestInfo={1}";
                  logger.debug(MessageFormat.format(logFormat, tx, requestInfo));
              }
  
              emitTuples(requestInfo.getFuncArgs(), collector);
              this.idsMap.put(tx, requestInfo);
          }
      }
docs: {@inheritDoc}