Hi @All I am facing an issue while printing Datastream[Row], tried ClosableIterator

val processedDataStreamOfRow: CloseableIterator[_] = dsStreamOfRow.executeAndCollect()
    while (processedDataStreamOfRow.hasNext) {
      println(processedDataStreamOfRow.next)
    }

Any suggestions ?

What's the issue?

so it does not print anything out there. and if I use for loop I am getting.. Failed to fetch next result java.lang.RuntimeException: Failed to fetch next result at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:109) at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80)

What is the rest of the exception stack trace that you're getting?

no I am using mini cluster to write a test. Failed to fetch next result java.lang.RuntimeException: Failed to fetch next result at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:109) at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80) at org.apache.flink.streaming.api.scala.CloseableIterator$$anon$1.hasNext(CloseableIterator.scala:36) at TestDefaultStreamPreProcessor$$anonfun$1.apply(TestDefaultStreamPreProcessor.scala:52) at TestDefaultStreamPreProcessor$$anonfun$1.apply(TestDefaultStreamPreProcessor.scala:24) at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85) at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104) at org.scalatest.Transformer.apply(Transformer.scala:22) at org.scalatest.Transformer.apply(Transformer.scala:20) at org.scalatest.funsuite.AnyFunSuiteLike$$anon$1.apply(AnyFunSuiteLike.scala:226) at org.scalatest.TestSuite$class.withFixture(TestSuite.scala:196) at org.scalatest.funsuite.AnyFunSuite.withFixture(AnyFunSuite.scala:1563) at org.scalatest.funsuite.AnyFunSuiteLike$class.invokeWithFixture$1(AnyFunSuiteLike.scala:223) at org.scalatest.funsuite.AnyFunSuiteLike$$anonfun$runTest$1.apply(AnyFunSuiteLike.scala:236) at org.scalatest.funsuite.AnyFunSuiteLike$$anonfun$runTest$1.apply(AnyFunSuiteLike.scala:236) at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306) at org.scalatest.funsuite.AnyFunSuiteLike$class.runTest(AnyFunSuiteLike.scala:236) at org.scalatest.funsuite.AnyFunSuite.runTest(AnyFunSuite.scala:1563) at org.scalatest.funsuite.AnyFunSuiteLike$$anonfun$runTests$1.apply(AnyFunSuiteLike.scala:269) at org.scalatest.funsuite.AnyFunSuiteLike$$anonfun$runTests$1.apply(AnyFunSuiteLike.scala:269) at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:413) at org.scalatest.SuperEngine$$anonfun$traverseSubNodes$1$1.apply(Engine.scala:401) at scala.collection.immutable.List.foreach(List.scala:392) at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401) at org.scalatest.SuperEngine.org$scalatest$SuperEngine$$runTestsInBranch(Engine.scala:396) at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475) at org.scalatest.funsuite.AnyFunSuiteLike$class.runTests(AnyFunSuiteLike.scala:269) at org.scalatest.funsuite.AnyFunSuite.runTests(AnyFunSuite.scala:1563) at org.scalatest.Suite$class.run(Suite.scala:1112) at org.scalatest.funsuite.AnyFunSuite.org$scalatest$funsuite$AnyFunSuiteLike$$super$run(AnyFunSuite.scala:1563) at org.scalatest.funsuite.AnyFunSuiteLike$$anonfun$run$1.apply(AnyFunSuiteLike.scala:273) at org.scalatest.funsuite.AnyFunSuiteLike$$anonfun$run$1.apply(AnyFunSuiteLike.scala:273) at org.scalatest.SuperEngine.runImpl(Engine.scala:535) at org.scalatest.funsuite.AnyFunSuiteLike$class.run(AnyFunSuiteLike.scala:273) at org.scalatest.funsuite.AnyFunSuite.run(AnyFunSuite.scala:1563) at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:45) at org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$1.apply(Runner.scala:1322) at org.scalatest.tools.Runner$$anonfun$doRunRunRunDaDoRunRun$1.apply(Runner.scala:1316) at scala.collection.immutable.List.foreach(List.scala:392) at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:1316) at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:972) at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:971) at org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:1482) at org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:971) at org.scalatest.tools.Runner$.run(Runner.scala:798) at org.scalatest.tools.Runner.run(Runner.scala) at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.runScalaTest2or3(ScalaTestRunner.java:38) at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.main(ScalaTestRunner.java:25) Caused by: java.io.IOException: Failed to fetch job execution result at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:177) at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:120) at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106) ... 47 more Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.client.JobExecutionException: Job execution failed. at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357) at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915) at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:175) ... 49 more Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.

As the exception says the job failed for some reason; you'll have to look into the logs to see what happened.

it says : No operators defined in streaming topology. Cannot execute.

Can you show the rest of the job?

class TestPreProcessor extends BaseUnit { val unmodifiableCollection = Class.forName("java.util.Collections$UnmodifiableCollection") val env = StreamExecutionEnvironment.getExecutionEnvironment env.addDefaultKryoSerializer(unmodifiableCollection, classOf[UnmodifiableCollectionsSerializer]) test("Test PreProcessor") { val demoService = spy(new DemoService(Array("--env", "test"))) demoService.getProvidedEnvironmentProperties("test", envProperties) val inputStreamString = "{\"@data\":\"{\\\"timestamp\\\":1665139067558,\\\"measure\\\":\\\"drive_usage\\\",\\\"tags\\\":{},\\\"area\\\":{\\\"area_name\\\":\\\"ny\\\",\\\"area_id\\\":\\\"A-8e19\\\"}}}" val inputStream: DataStream[_] = StreamExecutionEnvironment.getExecutionEnvironment.fromElements(inputStreamString) val preProcessedRow = demoService.preProcess(inputStream, config, env) // return a datastream of Row print(preProcessedRow) /*val preTest: CloseableIterator[_] = preProcessedRow.executeAndCollect() while (preTest.hasNext) { println(preTest.next) } val collection = DataStreamUtils.collect(preProcessedRow.javaStream) val results: Seq[_] = collection.asScala.toList for (result <- results){ print(result) }*/ env.execute("Main Job") } }

What does

demoService.preProcess

do?

it process DataStream of String and returns a DataStream of Row which is working file, only challenge here is how to get row values for assertion

You can't use

executeAndCollect

and

DataStreamUtils.collect

in the same job.

oh, do we have any reference for iterating over DataStream[_]