Udith

let AWS = require('aws-sdk');
const s3 = new AWS.S3();
exports.handler = function (event, context, callback) {

	console.log(`Batch process triggered at ${event.time}`);

	s3.listObjects({
		'Bucket': 'batch-process-bucket-udith',
		'MaxKeys': 100,
		'Prefix': ''

Udith

let AWS = require('aws-sdk');
const sns = new AWS.SNS();
const s3 = new AWS.S3();

const S3_BUCKET_NAME = 'batch-process-bucket-udith';
const SNS_TOPIC_ARN = 'arn:aws:sns:us-east-1:<YOUR_ACCOUNT_ID>:BatchProcessNotifications';

/* 
This is the main function of the lambda 
*/

Udith

@Override
public boolean add(Runnable runnable) {
    return super.offer(runnable);
}

Udith

public class ReEnqueuePolicy implements RejectedExecutionHandler {

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (!executor.getQueue().add(r)) {
            rejectedExecutionHandler.rejectedExecution(r, executor);
        }
    }
}

Udith

@Override
public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
    currentIdleThreadCount.incrementAndGet();
    Runnable poll = super.poll(timeout, unit);
    currentIdleThreadCount.decrementAndGet();
    return poll;
}

Udith

@Override
public Runnable take() throws InterruptedException {
    currentIdleThreadCount.incrementAndGet();
    Runnable take = super.take();
    currentIdleThreadCount.decrementAndGet();
    return take;
}

Udith

Runnable r = timed ? 
  workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();

Udith

@Override
public boolean offer(Runnable e) {
    return currentIdleThreadCount.get() > 0 && super.offer(e);
}

Udith

int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
    if (addWorker(command, true))
        return;
    c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
    int recheck = ctl.get();
    if (! isRunning(recheck) && remove(command))
        reject(command);

Udith

private void optimizeIndex() {
    OptimizeResponse optimizeResponse = elasticsearch.admin().indices()
            .prepareOptimize(ES_INDEX)
            .setFlush(true)
            .setOnlyExpungeDeletes(false)
            .execute().actionGet();

    System.out.println("Elasticsearch index optimization finished with " +
            optimizeResponse.getSuccessfulShards() + " successful and " +
            optimizeResponse.getFailedShards() + " failed shards out of " +