Flink Spector throws please specify the TypeInformation manually via StreamExecutionEnvironment#fromElements(Collection, TypeInformation) error about flink-spector HOT 5 CLOSED

Hi,

You are using scala which is not yet officially supported, but shouldn't pose any problems here as far I can see, since you're using the java api of Flink. You're indeed right that I should provide a way here to provide Typeinformation. I will try to include this in the next release.

But I think your problem is right here:
createTimedTestStreamWith(Tuple2.of(null, null))

I use the first element later to extract TypeInformation. I think your test would work fine if the first element in your TestStream / Input would contain actual data.

You could change the logic to provide the input like this:

var testStream :EventTimeSourceBuilder[Tuple2[String, Report]] = null

val it = ReportList.iterator()
var report:Tuple2[String,Report] = null
report = it.next()

testStream = createTimedTestStreamWith(Tuple2.of(report.f0, report.f1))

while(it.hasNext)
  {
    report = it.next()
    testStream = testStream.emit(Tuple2.of(report.f0, report.f1))

  }

This way the first element contains actual data.

And since your'e not providing any timestamps but do Windowing do a .closeAndFlush(); instead of a simple close() this will instruct Flinkspector to emit a max watermark at the end of the test stream.
val testStream1 = testStream.closeAndFlush();

I have not tested any of this. Please provide some Feedback if this resolves your problem.

Cheers,
Alex

from flink-spector.

Hello,

I have found somewhat a solution that works:

var builder: EventTimeSourceBuilder[Tuple2[String, Report]] = null
val reports: List[Tuple2[String, Report]] = ...

if (reports.size < 1) {
      assert(false)
}

builder = createTimedTestStreamWith(reports.head)
for (report <- reports.drop(1)) {
  builder.emit(report, InWindow.to( ... ))
}
val inputStream = builder.close()

The only issue I see is that the first record isn't inserted into a window, so I am not sure where the first record in the list ends up. Maybe #createTimedTestStreamWith could be updated so you can pass a time for the first record, or like previously suggested, just passing the type with the stream constructor, and emiting all the records like normal.

from flink-spector.

We got flink-spector working with scala driver by adding following in the sbt

"junit" % "junit" % "4.11" % Test,
crossPaths := false,

Implementation is described as above by chris.

class Consumer extends DataStreamTestBase with AssertionsForJUnit{
@test def testStream() = {
var builder: EventTimeSourceBuilder[Tuple2[String, Report]] = null
val reports: List[Tuple2[String, Report]] = ...

if (reports.size < 1) {
assert(false)
}

builder = createTimedTestStreamWith(reports.head)
for (report <- reports.drop(1)) {
builder.emit(report, InWindow.to( ... ))
}
val inputStream = builder.close()

}

}

you can run the test with "sbt test"

from flink-spector.

@cslotterback You can already to this by using the underlying EventTimeInputBuilder.

    public static <T> EventTimeInputBuilder<T> startWith(T record, long timeStamp) 

    public static <T> EventTimeInputBuilder<T> startWith(T record, Moment moment) 

But I'm going to expose this in the high level api in the next release.

@rohanbolar If I'm not mistaken according to the inputs your still using the java api of flink?
import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSource, SingleOutputStreamOperator}

from flink-spector.

This issue has been bundled into #57

from flink-spector.