Hello again, everybody. I’m using the Onyx SQS plugin with Onyx 0.10.0-beta17. I’m running into an issue when creating adding an SQS input task to my job. It’s claiming that I have keywords that are namespaced with things other than :onyx or :sqs, but my task map looks like this:

{:sqs/queue-name "my queue"
:onyx/plugin :onyx.plugin.sqs-input/input
:onyx/medium :sqs
:onyx/batch-timeout 1000,
:onyx/type :input
:sqs/attribute-names []
:onyx/name :in
:sqs/message-attribute-names []
:sqs/deserializer-fn :clojure.edn/read-string
:sqs/idle-backoff-ms 1000
:onyx/doc "Reads segments from an SQS queue"
:sqs/region "us-east-1"
:onyx/batch-size 2
}

The SQS plugin has this schema definition:

(def SQSInputTaskMap
  {(s/optional-key :sqs/queue-name) s/Str
   (s/optional-key :sqs/queue-url) s/Str
   (s/optional-key :onyx/batch-timeout) batch-timeout-check
   (s/optional-key :sqs/attribute-names) [s/Str]
   (s/optional-key :sqs/message-attribute-names) [s/Str]
   :sqs/region s/Str
   :onyx/batch-size max-batch-size
   :sqs/deserializer-fn os/NamespacedKeyword
   (os/restricted-ns :sqs) s/Any})

@stephenmhopper Hmm, it shouldn’t.

It should only be restricting sqs/ keys

Oh, does that mean that I have an extraneous :sqs/something entry?

@stephenmhopper I’d have to see the stacktrace, but that’s my guess, yeah

Is :sqs/deserializer-fn a thing?

Its been a long time since I’ve looked at the plugin

Ah, the :sqs/idle-backoff-ms entry shouldn’t be there. I removed it, and now it works. It was confusing because the error message seemed to indicate that all of my entries had to be namespaced with either :sqs or :onyx, so I was looking for the wrong thing

thank you!

It looks like :sqs/idle-backoff-ms is still in the readme even though the schema doesn’t allow it, so I’ve submitted an issue on GitHub to update one or the other.

@stephenmhopper Thanks! We’ll fix that up. Yeah, Schema’s error output isn’t great. 😕 Looking forward to switch it all out to Spec soon

That’s great to hear. I recently started switching all of my apps over from Schema to Spec and I love it

@stephenmhopper thanks, it was an issue in the docs

I’ve removed it from the README

@lucasbradstreet cool, thank you

@lucasbradstreet Does the Amazon SQS plugin have some way to delete messages after they’ve been successfully processed? Right now, it looks like if I write a message to a queue and then set up a separate job to read that message, it will just continually read that message. The job I’ve set up just goes :sqs-input -> :identity -> :core.async-out

Do I need to add another task for deleting the message part way through my job?

hmm, might be a bug with the 0.10 port

it should automatically clear it once it’s acknowledged

The logs just contain INFO statements about checkpointing. It’ll have six consecutive “Checkpointed input” statement followed by a single “Checkpointed output” statement.

@lucasbradstreet Is there something else I should be looking at to help troubleshoot this one?

Does Onyx handle the case where, I have some input from :input-1 that goes to :process-1.

Then from there, we have 1 immediate output to :output-1.

And a continuous stream of data to output-2?

[[:input-1 :process-1]
 [:process-1 :output-1]
 [:process-1 :output-2]]

@stephenmhopper looking into it myself. Will be in touch.

@twashing output-1 and output-2 will receive equivalent streams there

process-1 will message to both of them

@lucasbradstreet Right, ok.

So then I need something that consumes process-1 outputs to output-1 … then an asynchronous long-running process that writes to output-2.

Oh you’re asking if we have something to do that. Hmm

Right.

Not sure what the use case is there. That might help

You might be able to do it with a trigger and trigger/emit

I have a feeling that that long-running stream simply writes to kafka (in this case), outside of Onyx.

Umm, the use case is that a command kicks off a subscription to a long-running feed.

But I think I can just use Onyx to issue the command and get the command result.

… that long-running feed could just be streamed to Kafka directly.

Ah, I get it

Yeah, not sure how that would best be performed. I’m sure you can do it but it might be a little awkward. A custom trigger would do it for sure.

@stephenmhopper found the bug in onyx sqs. Will have a fix for you shortly

@lucasbradstreet cool, thank you! I just confirmed that if I delete the message manually on output that the problem goes away, but that was definitely a hack. What was the issue?

It looked like both checkpoint and checkpointed! weren’t doing much of anything. Was that it?

@lucasbradstreet Ok, I’ll dig into it. Thanks.

synced should have been deleting the messages for that epoch, but epoch was a volatile, so needed to be derefed

technically the call should be in checkpointed!, but onyx isn’t calling it yet.

I’ll try to get that happening soon

I’ll have a snapshot build for you shortly

this plugin wasn’t super well tested since 0.10

it’s been used quite a lot with 0.9 though

Ah, so was the (get @processing epoch) bit just returning nil?

yea

the test should really drain the input queue to make sure everything was deleted

If you’d like to give me a PR for that it’d be appreciated. I don’t have the time atm

A PR for the test or for the epoch thing?

or both?

test

Sure, I’ll see if I can do something with it within the next week

basically drain the input queue in here https://github.com/onyx-platform/onyx-amazon-sqs/blob/0.10.x/test/onyx/plugin/sqs_input_test.clj#L85 to make sure it doesn’t have anything on it

actually I might as well just do it now. don’t worry about it

okay, thanks again!