@Override
    public void process(Exchange exchange) throws Exception {
        Map<String, Object> headers = exchange.getIn().getHeaders();

        String id = (headers.containsKey(HEADER_ID)) ? exchange.getIn().getHeader(HEADER_ID, String.class) : endpoint.getId();

        int ttl = (headers.containsKey(HEADER_TTL))
                ? Integer.parseInt(exchange.getIn().getHeader(HEADER_TTL, String.class)) : DEFAULT_TTL;

        if (endpoint.isAutoStartIdForInserts()) {
            id = Long.toString(startId.getAndIncrement());
        } else if (id == null) {
            throw new CouchbaseException(HEADER_ID + " is not specified in message header or endpoint URL.", exchange);
        }

        if (endpoint.getOperation().equals(COUCHBASE_PUT)) {
            LOG.trace("Type of operation: PUT");
            Object obj = exchange.getIn().getBody();
            Boolean result = CouchbaseCollectionOperation.setDocument(collection, id, ttl, obj, persistTo, replicateTo,
                    writeQueryTimeout, producerRetryPause);
            exchange.getMessage().setBody(result);
        } else if (endpoint.getOperation().equals(COUCHBASE_GET)) {
            LOG.trace("Type of operation: GET");
            GetResult result = CouchbaseCollectionOperation.getDocument(collection, id, queryTimeout);
            exchange.getMessage().setBody(result);
        } else if (endpoint.getOperation().equals(COUCHBASE_DELETE)) {
            LOG.trace("Type of operation: DELETE");
            MutationResult result
                    = CouchbaseCollectionOperation.removeDocument(collection, id, writeQueryTimeout, producerRetryPause);
            exchange.getMessage().setBody(result.toString());
        }
        // cleanup the cache headers
        exchange.getIn().removeHeader(HEADER_ID);
    }
@Test
    public void testExpiryTimeIsSet() throws Exception {
        // Mock out some headers so we can set an expiry
        int expiry = 5000;
        Map<String, Object> testHeaders = new HashMap<>();
        testHeaders.put("CCB_TTL", Integer.toString(expiry));
        when(msg.getHeaders()).thenReturn(testHeaders);
        when(collection.upsert(anyString(), any(), any())).thenReturn(response);
        when(msg.getHeader(HEADER_TTL, String.class)).thenReturn(Integer.toString(expiry));

        when(endpoint.getId()).thenReturn("123");
        when(endpoint.getOperation()).thenReturn("CCB_PUT");
        when(exchange.getMessage()).thenReturn(msg);
        ArgumentCaptor<UpsertOptions> options = ArgumentCaptor.forClass(UpsertOptions.class);

        producer.process(exchange);

        verify(collection).upsert(anyString(), any(), options.capture());
    }