Realized I also need to setup DistributedMapCacheClientService along with
DistributedMapCacheServer. [ wonder how many instances service will be running
in clustered env)
but still looks for guidelines on how to make
org.apache.nifi.distributed.cache.* available for ExecuteScript processor !
Thanks
Sumo
> On Jul 13, 2016, at 9:12 PM, Sumanth Chinthagunta <xmlk...@gmail.com> wrote:
>
> Matt,
> I setup DistributedMapCacheServer controller service and trying to run
> following script. Am I doing correct?
> I have couple of issues:
> 1. How do I make org.apache.nifi.distributed.cache.* package available for
> ExecuteScript?
> 2. When I try : cache.get("1",null,null), getting following error
> Failed to process session due to
> org.apache.nifi.processor.exception.ProcessException:
> javax.script.ScriptException: javax.script.ScriptException:
> groovy.lang.MissingMethodException: No signature of method:
> com.sun.proxy.$Proxy129.get() is applicable for argument types:
> (java.lang.String, null, null) values: [1, null, null]
> Possible solutions: grep(), wait(), any(), getAt(java.lang.String), dump(),
> find(): org.apache.nifi.processor.exception.ProcessException:
> javax.script.ScriptException: javax.script.ScriptException:
> groovy.lang.MissingMethodException: No signature of method:
> com.sun.proxy.$Proxy129.get() is applicable for argument types:
> (java.lang.String, null, null) values: [1, null, null]
> Possible solutions: grep(), wait(), any(), getAt(java.lang.String), dump(),
> find()
>
>
>
>
> import org.apache.nifi.controller.ControllerService
> import org.apache.nifi.distributed.cache.client.Deserializer;
> import org.apache.nifi.distributed.cache.client.Serializer;
> import
> org.apache.nifi.distributed.cache.client.exception.DeserializationException;
> import
> org.apache.nifi.distributed.cache.client.exception.SerializationException;
>
> static class StringSerializer implements Serializer<String> {
>
> @Override
> public void serialize(final String value, final OutputStream out) throws
> SerializationException, IOException {
> out.write(value.getBytes(StandardCharsets.UTF_8));
> }
> }
>
> static class CacheValueDeserializer implements Deserializer<byte[]> {
>
> @Override
> public byte[] deserialize(final byte[] input) throws
> DeserializationException, IOException {
> if (input == null || input.length == 0) {
> return null;
> }
> return input;
> }
> }
>
> private final Serializer<String> keySerializer = new StringSerializer();
> private final Deserializer<byte[]> valueDeserializer = new
> CacheValueDeserializer();
>
> def lookup = context.controllerServiceLookup
> def cacheServerName = distributedMapCacheServerName.value
>
> def cacheServerId =
> lookup.getControllerServiceIdentifiers(ControllerService).find {
> cs -> lookup.getControllerServiceName(cs) == cacheServerName
> }
>
> def cache = lookup.getControllerService(cacheServerId)
>
> //log.error cache.get("1",keySerializer,valueDeserializer)
> log.error cache.get("1",null,null)
>
> Thanks
> Sumo
>
>
>
>
>> On Jul 13, 2016, at 6:13 PM, Matt Burgess <mattyb...@gmail.com
>> <mailto:mattyb...@gmail.com>> wrote:
>>
>> Sumo,
>>
>> I have some example code at
>> http://funnifi.blogspot.com/2016/04/inspecting-your-nifi.html
>> <http://funnifi.blogspot.com/2016/04/inspecting-your-nifi.html>.
>> Although it's not expressly for ExecuteScript it should be pretty
>> usable as-is.
>>
>> Also you might be able to use the technique outlined in my other post:
>> http://funnifi.blogspot.com/2016/04/sql-in-nifi-with-executescript.html
>> <http://funnifi.blogspot.com/2016/04/sql-in-nifi-with-executescript.html>.
>> With this you would get a reference to the ControllerService for the
>> DistributedMapCacheClient (although you likely won't be able to refer
>> to the DistributedMapCacheClient class), but using dynamic method
>> invocation (as outlined in the blog post) you can call its methods to
>> get and put values.
>>
>> Regards,
>> Matt
>>
>> On Wed, Jul 13, 2016 at 8:26 PM, Sumanth Chinthagunta <xmlk...@gmail.com
>> <mailto:xmlk...@gmail.com>> wrote:
>>> looking for example script ( for ExecuteScript processor) that uses
>>> DistributedMapCacheClient to put/get key/value pair.
>>>
>>> Thanks
>>> -Sumo
>
