vmarcinko

apply plugin: 'java'

archivesBaseName = 'bigdatalab'
group = 'bigdatalab'
version = '1.0'

ext {
    slf4jVersion = '1.7.5'
    hadoopVersion = '1.0.4'
    cascadingVersion = '2.1.+'

vmarcinko

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.mapreduce.JobSubmissionFiles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;

vmarcinko

import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

vmarcinko

public class CategoryEnumInfo {
    private final String label;
    private final Byte encodedValue;

    public CategoryEnumInfo(String label, Byte encodedValue) {
        this.label = label;
        this.encodedValue = encodedValue;
    }

    public String getLabel() {

vmarcinko

package vmarcinko.undertow;

import io.undertow.Handlers;
import io.undertow.Undertow;
import io.undertow.server.HttpHandler;
import io.undertow.server.handlers.PathHandler;
import io.undertow.server.handlers.RedirectHandler;
import io.undertow.server.handlers.resource.FileResourceManager;
import io.undertow.servlet.Servlets;
import io.undertow.servlet.api.DeploymentInfo;

vmarcinko

@Embeddable
public class QueueingState {
  
	public enum Status {
		NOT_ATTEMPTED,
		ERROR,
		SUCCESS
	}

	private Status status;

vmarcinko

public interface QueueConsumerModule<ID> {
   List<ID> findItemIdsWhereQueueingNextAttemptTimeIsBefore(LocalDateTime time, int limit);
   Optional<QueueingState> getQueueingStateForItem(ID itemId);
   Optional<QueueingState> processItem(ID itemId);
}

vmarcinko

private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
// ...
private void startProcessingTask() {
   logger.info("Starting queue processing task with delay of {} secs", this.pollingPeriodInSecs);
   Runnable command = this::processQueuedItems;
   this.processingTask = this.scheduledExecutorService.scheduleWithFixedDelay(command, pollingPeriodInSecs, pollingPeriodInSecs, TimeUnit.SECONDS);
}

vmarcinko

public void processQueuedItems() {
	try {
		LocalDateTime now = LocalDateTime.now();
		List<?> itemIds = this.queueConsumerModule.findItemIdsWhereQueueingNextAttemptTimeIsBefore(now, itemsPollSize);

		if (!itemIds.isEmpty()) {
			logger.info("Fetched {} pending queued items", itemIds.size());
			for (Object itemId : itemIds) {
				processItemAndHandleErrorIfRaised(itemId);
			}

vmarcinko

@Entity
@Table(name = "sms_message", indexes = @Index(name = "idx_sms_msg_queue_polling_fields", columnList = "next_attempt_time"))
public class SmsMessage {
	@Id
	@Column(name = "id", nullable = false)
	@GeneratedValue
	private Long id;

	@Column(name = "uid", nullable = false, unique = true)
	private String uid; // app-assigned unique ID