Data Transfer: Consumer

Consumers are used to send the track requests to the Intelligence track server. There are ...

• HTTP_CLIENT: Saves track requests in memory and sends them to the server using separate parallel threads.

• FILE: Requests are saved in a separate file and send via a cronjob to the track server in a specified interval.

• CUSTOM: Create your own consumer to send the tracking requests.

The requests are collected in in-memory queues. The queues are sent when flush() is called, every requestInterval seconds if an interval is configured, and when the tracking instance is closed (see Lifecycle and Threading). What happens with a batch is decided by the consumer:

Note

Requests that are still in the memory queue when the tracking instance is closed get one last flush with up to maxAttempt attempts, within the 30-second deadline of close(). Whatever is still undelivered after that is lost. Each queue is bounded by maxQueueSize: when a queue reaches it, the queue is sent; if that flush fails, the oldest requests are discarded (logged at level WARN). 0 means unlimited.

MappIntelligenceConfig config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.HTTP_CLIENT)
    .setMaxAttempt(1)
    .setAttemptTimeout(100)
    .setMaxBatchSize(1)
	.setMaxQueueSize(1000)
    .setRequestInterval(0)
    .setRequestQueues(2);

MappIntelligenceConfig config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.FILE)
	.setFilePath("/path/to/directory/")
	.setFilePrefix("MappIntelligenceRequests")
	.setMaxAttempt(1)
    .setAttemptTimeout(100)
    .setMaxBatchSize(50) // ignored for the FILE consumer, always 1
	.setMaxFileLines(5000)
	.setMaxFileDuration(30 * 60 * 1000)
	.setMaxFileSize(12 * 1024 * 1024);

class TestCustomConsumer implements MappIntelligenceConsumer {
	public static final String CUSTOM = "CUSTOM";
	public boolean sendBatch(List<String> batchContent) {
		// implement custom consumer logic
		return true;
	}
}
MappIntelligenceConfig config = new MappIntelligenceConfig()
	.setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
	.setConsumerType(MappIntelligenceConsumerType.CUSTOM)
	.setConsumer(new TestCustomConsumer());

class TestCustomConsumer implements MappIntelligenceConsumer {
    public boolean sendBatch(List<String> batchContent) {
        return this.send(batchContent) == BatchResult.SENT;
    }

    @Override
    public BatchResult send(List<String> batchContent) {
        // implement custom consumer logic
        // return BatchResult.SENT, BatchResult.RETRY or BatchResult.DISCARDED
        return BatchResult.SENT;
    }
}

Examples for custom consumers

The following examples show CUSTOM consumers for two typical use cases: buffering the requests in Kafka or Redis before they are sent to the track server, and checking your tracking in unit tests. The examples are not part of the library. Adapt them to your environment.

Kafka

Writes every request as one message into a Kafka topic (client library org.apache.kafka:kafka-clients). KafkaProducer is thread-safe, so one producer can be shared by all queue threads and tracking instances. The batch counts as SENT only when the broker has confirmed every message; if a message fails, the whole batch is RETRY.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

class KafkaTrackingConsumer implements MappIntelligenceConsumer {
    private final Producer<String, String> producer;
    private final String topic;

    KafkaTrackingConsumer(Producer<String, String> producer, String topic) {
        this.producer = producer;
        this.topic = topic;
    }

    public boolean sendBatch(List<String> batchContent) {
        List<Future<RecordMetadata>> results = new ArrayList<>(batchContent.size());
        for (String request : batchContent) {
            results.add(this.producer.send(new ProducerRecord<>(this.topic, request)));
        }

        try {
            for (Future<RecordMetadata> result : results) {
                result.get(10, TimeUnit.SECONDS);
            }
            return true;
        } catch (ExecutionException | TimeoutException e) {
            // broker not reachable: the batch goes back into the queue
            return false;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");
// one producer per application, close it yourself when the application stops
Producer<String, String> producer = new KafkaProducer<>(props);

MappIntelligenceConfig config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.CUSTOM)
    .setConsumer(new KafkaTrackingConsumer(producer, "mapp-intelligence-requests"));

A KafkaException thrown by send() (for example when the producer buffer is full) is a RuntimeException, so the library logs it and treats the batch as RETRY.

Redis

Appends the requests to a Redis list (client library redis.clients:jedis). JedisPool is thread-safe; every call takes its own connection from the pool. RPUSH with all requests of the batch is one command, so the batch is appended completely or not at all.

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.exceptions.JedisException;

class RedisTrackingConsumer implements MappIntelligenceConsumer {
    private final JedisPool pool;
    private final String key;

    RedisTrackingConsumer(JedisPool pool, String key) {
        this.pool = pool;
        this.key = key;
    }

    public boolean sendBatch(List<String> batchContent) {
        try (Jedis jedis = this.pool.getResource()) {
            jedis.rpush(this.key, batchContent.toArray(new String[0]));
            return true;
        } catch (JedisException e) {
            // Redis not reachable: the batch goes back into the queue
            return false;
        }
    }
}

// one pool per application, close it yourself when the application stops
JedisPool pool = new JedisPool("redis", 6379);

MappIntelligenceConfig config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.CUSTOM)
    .setConsumer(new RedisTrackingConsumer(pool, "mapp-intelligence-requests"));

Forwarding the requests from Kafka or Redis

Kafka and Redis only buffer the requests. A separate process of your own (a worker) has to read them and send them to the track server, the same way the HTTP_CLIENT consumer does:

• POST to https://<trackDomain>/<trackId>/batch with the content type text/plain, one request per line in the body.

• At most 10,000 requests and less than 24 MB per batch.

• Handle the response like the HTTP_CLIENT consumer (see the table of batch results above): a status below 400 means delivered; 5xx, 408, 429 and network errors are retried; any other 4xx can never be delivered.

• Commit the Kafka offset, or remove the requests from the Redis list, only after the batch was delivered. With Redis and one worker per list, for example read with LRANGE <key> 0 9999 and after delivery remove the batch with LTRIM <key> <number of requests> -1.

Note

If a batch is retried after part of it already reached Kafka or Redis (for example a timeout after the broker has stored the message), these requests are buffered twice. Kafka and Redis do not prevent this, and neither does the worker. If duplicates matter, deduplicate in the worker.

Test consumer

Keeps the requests in memory instead of sending them, so you can check in your own unit tests what the library would track, without network or files. setAvailable(false) simulates an unreachable track server: every batch is then RETRY and stays in the queue.

class InMemoryTestConsumer implements MappIntelligenceConsumer {
    // thread-safe, the consumer is called from all queue threads
    private final List<String> requests = new CopyOnWriteArrayList<>();
    private volatile boolean available = true;

    public boolean sendBatch(List<String> batchContent) {
        if (!this.available) {
            return false;
        }

        this.requests.addAll(batchContent);
        return true;
    }

    List<String> getRequests() {
        return new ArrayList<>(this.requests);
    }

    void setAvailable(boolean available) {
        this.available = available;
    }

    void clear() {
        this.requests.clear();
    }
}

Usage in a JUnit test. flush() only hands the requests to the background threads, so close the instance before you check the requests: close() waits until the remaining requests have been handed to the consumer.

@Test
public void testTrackPage() {
    InMemoryTestConsumer consumer = new InMemoryTestConsumer();
    MappIntelligenceConfig config = new MappIntelligenceConfig()
        .setTrackDomain("analytics01.wt-eu02.net")
        .setTrackId("111111111111111")
        .setConsumerType(MappIntelligenceConsumerType.CUSTOM)
        .setConsumer(consumer);

    try (MappIntelligenceTracking mappIntelligence = new MappIntelligenceTracking(config)) {
        mappIntelligence.track(new MappIntelligenceDataMap().page(new MappIntelligencePage("en.home")));
    }

    assertEquals(1, consumer.getRequests().size());
    assertTrue(consumer.getRequests().get(0).contains("en.home"));
}

Please check Configuration to see all configuration options and explanations of the Mapp Intelligence Java library, including default settings.