Aerospike Abnormal Increase In Memory Consumption when using UDF via Java Client (Query Aggregation)
<p>So let me explain first. I have a sample snippet of code from Java that is written like this:</p>

<pre><code>Statement statement = new Statement();
statement.setNamespace("foo");
statement.setSetName("bar");
statement.setAggregateFunction(Thread.currentThread().getContextClassLoader(),
    "udf/resource/path","udfFilename","udfFunctionName",
    "args1","args2","args3","args4");
ResultSet rs = aerospikeClient.getClient().queryAggregate(null,statement);

while(rs.next()){
   //insert logic code here 
}
</code></pre>

<p>With that snippet of sample code, I was able to use a UDF written in lua as sampled by the Aerospike Documentation. The UDF just search around the multiple bins and returns back its findings, it never persist nor transforms any data.</p>

<p>Now the thing is, when the function that uses this code that invokes UDF, in AMC (Aerospike Management Console) it spawns aggregation jobs that is mark as "done(ok)" but never mark as completed, and still are under the "Running Jobs" Table not the "Completed Jobs" Table. (see picture below)</p>

<p><a href="https://i.stack.imgur.com/Riq7T.png" rel="nofollow">Jobs Under Running Jobs Table</a></p>

<p><a href="https://i.stack.imgur.com/UMzHd.png" rel="nofollow">Jobs Under Completed Jos Table</a></p>

<p>and under Bash Terminal command "Top" I saw the Memory Percentage Usage of the Aerospike Server just keeps growing as the Jobs grew even more in number until such a time the Aerospike Server Fails since it maxed out its memory usage of the machine.</p>

<p>My questions are, </p>

<ol>
<li>is it possible for the jobs to let go of this resources (if they are really what transpire on the abnormal memory increase)?</li>
<li>if it wasn't the jobs that is responsible, what is?</li>
</ol>

<p>***EDITED:
Sample Lua Code:</p>

<pre><code>local function map_request(record)
    return map {response = record.response,
                templateId = record.templateId, 
                id = record.id, requestSent = record.requestSent,
                dateReplied = record.dateReplied}
end

function checkResponse(stream, responseFilter, templateId, validtyPeriod, currentDate)
       local function filterResponse(record)
            if responseFilter ~= "FOO" and validityPeriod &gt; 0 then
                   return (record.response == responseFilter) and 
                          (record.templateId == templateId) and 
                          (record.dateReplied + validityPeriod) &gt; currentDate
            else
                   return (record.response == responseFilter) and 
                          (record.templateId == templateId)
            end
        end
        return stream:filter (filterResponse):map(map_request)
end
</code></pre>