bulk-cdk: improve AirbyteConnectorRunner and CliRunner

[postamar]

What

The CliRunner dumps all output into a temp file and re-reads it into a BufferingOutputConsumer. This works, but it requires an edge case in the CLI grammar defined in AirbyteConnectorRunner. @edgao found a much better way in #45113 involving RuntimeBeanDefinitions. This PR implements this change and removes the --output CLI argument support. It also proves a nice entry point for destination input messages in the CliRunner.

How

The CliRunner overrides the PrintStream that gets injected in the StdoutOutputConsumer. By default it's a PrintStream which wraps System.out but for integration tests we want the StdoutOutputConsumer to feed into a BufferingOutputConsumer.

Same goes for InputStream for destinations.

Review guide

n/a

User Impact

None, test-only change.

Can this PR be safely reverted and rolled back?

• YES 💚
• NO ❌

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Sep 11, 2024 0:45am

[edgao]

had a question in CliRunner about how to use this interface, plus a few nits

[postamar]

This is ready for another look. I added the missing functionality for async running.

[edgao]

checking my understanding: there's two ways to use this asyncly:

val dest = CliRunner.destination(pipeInput)
launch { dest.run() }
sendMessages(pipe)
inspect(dest.results.stuff())

or via callback

CliRunner.destination(byteArrayInput)
  .withCallback({ msg -> ... })
  .run()

lgtm assuming I'm on the right track!

[postamar]

Yes indeed! The callback is required if you want to interrupt the sync sooner than normal operation.

Something which I didn't explore (because I felt it was out of scope for this PR) but which would make sense is to use channels instead of a PipeInputStream for destination.