title: Number of input rows in spark structured streaming with custom sink
body: <p>I'm using a custom sink in structured stream (spark 2.2.0) and noticed that spark produces incorrect metrics for number of input rows - it's always zero.</p>
  
  <p>My stream construction:</p>
  
  <pre><code>StreamingQuery writeStream = session
              .readStream()
              .schema(RecordSchema.fromClass(TestRecord.class))
              .option(OPTION_KEY_DELIMITER, OPTION_VALUE_DELIMITER_TAB)
              .option(OPTION_KEY_QUOTE, OPTION_VALUE_QUOTATION_OFF)
              .csv(s3Path.toString())
              .as(Encoders.bean(TestRecord.class))
              .flatMap(
                  ((FlatMapFunction&lt;TestRecord, TestOutputRecord&gt;) (u) -&gt; {
                      List&lt;TestOutputRecord&gt; list = new ArrayList&lt;&gt;();
                      try {
                          TestOutputRecord result = transformer.convert(u);
                          list.add(result);
                      } catch (Throwable t) {
                          System.err.println("Failed to convert a record");
                          t.printStackTrace();
                      }
  
                      return list.iterator();
                  }),
                  Encoders.bean(TestOutputRecord.class))
          .map(new DataReinforcementMapFunction&lt;&gt;(), Encoders.bean(TestOutputRecord.clazz))
          .writeStream()
          .trigger(Trigger.ProcessingTime(WRITE_FREQUENCY, TimeUnit.SECONDS))
          .format(MY_WRITER_FORMAT)
          .outputMode(OutputMode.Append())
          .queryName("custom-sink-stream")
          .start();
  
          writeStream.processAllAvailable();
          writeStream.stop();
  </code></pre>
  
  <p>Logs:</p>
  
  <pre><code>Streaming query made progress: {
    "id" : "a8a7fbc2-0f06-4197-a99a-114abae24964",
    "runId" : "bebc8a0c-d3b2-4fd6-8710-78223a88edc7",
    "name" : "custom-sink-stream",
    "timestamp" : "2018-01-25T18:39:52.949Z",
    "numInputRows" : 0,
    "inputRowsPerSecond" : 0.0,
    "processedRowsPerSecond" : 0.0,
    "durationMs" : {
      "getOffset" : 781,
      "triggerExecution" : 781
    },
    "stateOperators" : [ ],
    "sources" : [ {
      "description" : "FileStreamSource[s3n://test-bucket/test]",
      "startOffset" : {
        "logOffset" : 0
      },
      "endOffset" : {
        "logOffset" : 0
      },
      "numInputRows" : 0,
      "inputRowsPerSecond" : 0.0,
      "processedRowsPerSecond" : 0.0
    } ],
    "sink" : {
      "description" : "com.mycompany.spark.MySink@f82a99"
    }
  }
  </code></pre>
  
  <p>Do I have to populate any metrics in my custom sink to be able to track progress? Or could it be a problem in FileStreamSource when it reads from s3 bucket?</p>