Hello Dev team,
We are giving our first shot in writing Beam IO connector for Oracle
Streaming Service (OSS). The plan is to first implement it for enterprise
use and based on the feedback and stability make it available open source.
This is our first attempt in developing a Beam IO connector and so far we
have progressed with the help of Beam documentation and other related IOs
like KafkaIO, KinesisIO. Thanks to the community on that front.
Now OSS *has a read limit of 200ms* so when we read the data as shown below
in our UnboundedReaders *advance()* method
// Get Messages
GetMessagesResponse getResponse = this.streamClient.getMessages(getRequest);
We are able to read around five message but after that we are getting *request
throttling error*
Request was throttled because requests limit exhausted, next request can be
made in 200 ms
We tried with an initial solution of introducing *Thread.sleep(200)* before
the getMessages to see how it is behaving and this time we are *able to
read around 300+ messages*. With the expertise available in this forum, I
would like to hear inputs on two points.
1.
How to implement this feature in a proper way rather than just with a
one-line Thread.sleep(200)
2.
After adding Thread.sleep(200) and reading 300+ messages the pipeline
encountered below error. I do not see any implementation specific detail in
the stack trace. Can I get an insight what this error could be and how to
handle.
java.lang.NullPointerException
at org.apache.beam.sdk.util.CoderUtils.encodeToSafeStream
(CoderUtils.java:82)
at org.apache.beam.sdk.util.CoderUtils.encodeToByteArray
(CoderUtils.java:66)
at org.apache.beam.sdk.util.CoderUtils.encodeToByteArray
(CoderUtils.java:51)
at org.apache.beam.sdk.util.CoderUtils.clone (CoderUtils.java:141)
at
org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.getReader
(UnboundedReadEvaluatorFactory.java:224)
at
org.apache.beam.runners.direct.UnboundedReadEvaluatorFactory$UnboundedReadEvaluator.processElement
(UnboundedReadEvaluatorFactory.java:132)
at org.apache.beam.runners.direct.DirectTransformExecutor.processElements
(DirectTransformExecutor.java:160)
at org.apache.beam.runners.direct.DirectTransformExecutor.run
(DirectTransformExecutor.java:124)
at java.util.concurrent.Executors$RunnableAdapter.call
(Executors.java:511)
at java.util.concurrent.FutureTask.run (FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker
(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run
(ThreadPoolExecutor.java:624)
at java.lang.Thread.run (Thread.java:748)
--
Thanks,
Praveen K Viswanathan
