@Test
public void testBinaryDocument() throws Exception {

    Bucket bucket = mock(Bucket.class);
    String inFileDataStr = "doc-in";
    String content = "binary";
    ByteBuf buf = Unpooled.copiedBuffer(content.getBytes(StandardCharsets.UTF_8));
    when(bucket.get(inFileDataStr, BinaryDocument.class))
        .thenReturn(BinaryDocument.create(inFileDataStr, buf));
    setupMockBucket(bucket);


    byte[] inFileData = inFileDataStr.getBytes(StandardCharsets.UTF_8);
    testRunner.enqueue(inFileData);
    testRunner.setProperty(DOCUMENT_TYPE, DocumentType.Binary.toString());
    testRunner.run();

    testRunner.assertTransferCount(REL_SUCCESS, 1);
    testRunner.assertTransferCount(REL_ORIGINAL, 1);
    testRunner.assertTransferCount(REL_RETRY, 0);
    testRunner.assertTransferCount(REL_FAILURE, 0);
    MockFlowFile outFile = testRunner.getFlowFilesForRelationship(REL_SUCCESS).get(0);
    outFile.assertContentEquals(content);
    MockFlowFile orgFile = testRunner.getFlowFilesForRelationship(REL_ORIGINAL).get(0);
    orgFile.assertContentEquals(inFileDataStr);
}
 

protected void handleGetAsyncBinaryDocument(BinaryDocument binaryDocument, CompletableFuture<V> future, String key) {
  ByteBuf buffer = binaryDocument.content();
  try {
    byte[] bytes;
    if (buffer.hasArray() && buffer.arrayOffset() == 0 && buffer.readableBytes() == buffer.array().length) {
      bytes = buffer.array();
    } else {
      bytes = new byte[buffer.readableBytes()];
      buffer.readBytes(bytes);
    }
    future.complete(valueSerde.fromBytes(bytes));
  } catch (Exception e) {
    future.completeExceptionally(
        new SamzaException(String.format("Failed to deserialize value of key %s with given serde", key), e));
  } finally {
    ReferenceCountUtil.release(buffer);
  }
}
 

@Override
public boolean advance() throws IOException {
    while (true) {
        ByteBuf event;
        try {
            event = resultsQueue.poll(1, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            LOG.error("Failed to poll event from the results queue", e);
            return false;
        }
        if (event != null) {
            currentRecord = converter.convertToAvro(event);
            connection.acknowledge(event);
            event.release();
            recordCount++;
            return true;
        }
        if (!connection.isStreaming() && resultsQueue.isEmpty()) {
            break;
        }
    }
    return false;
}
 

public CouchbaseStreamingConnection(String bootstrapNodes, String bucket, String password) {
    connected = false;
    streaming = false;
    client = Client.configure()
            .connectTimeout(20000L)
            .hostnames(bootstrapNodes)
            .bucket(bucket)
            .password(password == null ? "" : password)
            .controlParam(DcpControl.Names.CONNECTION_BUFFER_SIZE, 20480)
            .bufferAckWatermark(60)
            .build();
    client.controlEventHandler(new ControlEventHandler() {

        @Override
        public void onEvent(ChannelFlowController controller, ByteBuf event) {
            controller.ack(event);
            event.release();
        }
    });

    dataEventHandler = new EventHandler();

    client.dataEventHandler(dataEventHandler);
}
 

@Override
public void onEvent(ChannelFlowController controller, ByteBuf event) {
    if (controller != null) {
        this.classController = controller;
    }
    if (resultsQueue != null) {
        try {
            resultsQueue.put(event);
        } catch (InterruptedException e) {
            LOG.error("Unable to put DCP request into the results queue");
        }
    } else {
        controller.ack(event);
        event.release();
    }
}
 

@Test
public void testStartStreaming() throws InterruptedException {
    Mockito.when(client.initializeState(StreamFrom.BEGINNING, StreamTo.NOW)).thenReturn(Completable.complete());
    Mockito.when(client.startStreaming(Mockito.<Short[]>anyVararg())).thenReturn(Completable.complete());
    SessionState sessionState = Mockito.mock(SessionState.class);
    Mockito.when(sessionState.isAtEnd()).thenReturn(false, false, true);
    Mockito.when(client.sessionState()).thenReturn(sessionState);

    BlockingQueue<ByteBuf> resultsQueue = new ArrayBlockingQueue<>(3);

    streamingConnection.startStreaming(resultsQueue);

    Assert.assertTrue(streamingConnection.isStreaming());

    Thread.sleep(2000);
    Mockito.verify(client, Mockito.times(3)).sessionState();
}
 

@Test
public void testStartStopStreaming() throws InterruptedException {
    Mockito.when(client.initializeState(StreamFrom.BEGINNING, StreamTo.NOW)).thenReturn(Completable.complete());
    Mockito.when(client.startStreaming(Mockito.<Short[]>anyVararg())).thenReturn(Completable.complete());
    Mockito.when(client.stopStreaming(Mockito.<Short[]>anyVararg())).thenReturn(Completable.complete());
    SessionState sessionState = Mockito.mock(SessionState.class);
    Mockito.when(sessionState.isAtEnd()).thenReturn(false, true);
    Mockito.when(client.sessionState()).thenReturn(sessionState);

    BlockingQueue<ByteBuf> resultsQueue = new ArrayBlockingQueue<>(4);
    resultsQueue.put(Mockito.mock(ByteBuf.class));
    resultsQueue.put(Mockito.mock(ByteBuf.class));
    resultsQueue.put(Mockito.mock(ByteBuf.class));
    resultsQueue.put(Mockito.mock(ByteBuf.class));
    Mockito.when(client.disconnect()).thenReturn(Completable.complete());


    streamingConnection.startStreaming(resultsQueue);
    streamingConnection.stopStreaming();

    Thread.sleep(1500);
    Mockito.verify(client, Mockito.times(2)).sessionState();
    Mockito.verify(client, Mockito.times(1)).disconnect();
}
 

@Test
public void testConvertToAvroDeletion() {
    ByteBuf buffer = Mockito.mock(ByteBuf.class);

    // Mocking key object
    Mockito.when(buffer.getByte(0)).thenReturn(MessageUtil.MAGIC_REQ);
    Mockito.when(buffer.getByte(1)).thenReturn(MessageUtil.DCP_DELETION_OPCODE);
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    ByteBuf key = Mockito.mock(ByteBuf.class);
    Mockito.when(key.readableBytes()).thenReturn(4);
    Mockito.when(buffer.slice(29, 10)).thenReturn(key);

    IndexedRecord record = converter.convertToAvro(buffer);

    assertIndexedRecordResult(record, "deletion", null);

}
 

@Test
public void testConvertToAvroExpiration() {
    ByteBuf buffer = Mockito.mock(ByteBuf.class);

    // Mocking key object
    Mockito.when(buffer.getByte(0)).thenReturn(MessageUtil.MAGIC_REQ);
    Mockito.when(buffer.getByte(1)).thenReturn(MessageUtil.DCP_EXPIRATION_OPCODE);
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    ByteBuf key = Mockito.mock(ByteBuf.class);
    Mockito.when(key.readableBytes()).thenReturn(4);
    Mockito.when(buffer.slice(29, 10)).thenReturn(key);

    IndexedRecord record = converter.convertToAvro(buffer);

    assertIndexedRecordResult(record, "expiration", null);

}
 

@Test (expected = IndexOutOfBoundsException.class)
public void testConvertToAvroFailedIllegalRecordIndex() {
    ByteBuf buffer = Mockito.mock(ByteBuf.class);

    // Mocking key object
    Mockito.when(buffer.getByte(0)).thenReturn(MessageUtil.MAGIC_REQ);
    Mockito.when(buffer.getByte(1)).thenReturn(MessageUtil.DCP_EXPIRATION_OPCODE);
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    ByteBuf key = Mockito.mock(ByteBuf.class);
    Mockito.when(key.readableBytes()).thenReturn(4);
    Mockito.when(buffer.slice(29, 10)).thenReturn(key);

    IndexedRecord record = converter.convertToAvro(buffer);

    record.get(10);
}
 

@Test (expected = UnmodifiableAdapterException.class)
public void testConvertToAvroFailedToChangeRecord() {
    ByteBuf buffer = Mockito.mock(ByteBuf.class);

    // Mocking key object
    Mockito.when(buffer.getByte(0)).thenReturn(MessageUtil.MAGIC_REQ);
    Mockito.when(buffer.getByte(1)).thenReturn(MessageUtil.DCP_EXPIRATION_OPCODE);
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    ByteBuf key = Mockito.mock(ByteBuf.class);
    Mockito.when(key.readableBytes()).thenReturn(4);
    Mockito.when(buffer.slice(29, 10)).thenReturn(key);

    IndexedRecord record = converter.convertToAvro(buffer);

    record.put(0, "Update is not supported");
}
 

@Test
public void testBinaryDocument() throws Exception {

    Bucket bucket = mock(Bucket.class);
    String inFileDataStr = "doc-in";
    String content = "binary";
    ByteBuf buf = Unpooled.copiedBuffer(content.getBytes(StandardCharsets.UTF_8));
    when(bucket.get(inFileDataStr, BinaryDocument.class))
        .thenReturn(BinaryDocument.create(inFileDataStr, buf));
    setupMockBucket(bucket);


    byte[] inFileData = inFileDataStr.getBytes(StandardCharsets.UTF_8);
    testRunner.enqueue(inFileData);
    testRunner.setProperty(DOCUMENT_TYPE, DocumentType.Binary.toString());
    testRunner.run();

    testRunner.assertTransferCount(REL_SUCCESS, 1);
    testRunner.assertTransferCount(REL_ORIGINAL, 1);
    testRunner.assertTransferCount(REL_RETRY, 0);
    testRunner.assertTransferCount(REL_FAILURE, 0);
    MockFlowFile outFile = testRunner.getFlowFilesForRelationship(REL_SUCCESS).get(0);
    outFile.assertContentEquals(content);
    MockFlowFile orgFile = testRunner.getFlowFilesForRelationship(REL_ORIGINAL).get(0);
    orgFile.assertContentEquals(inFileDataStr);
}
 

@Test
public void testBinaryDocumentToAttribute() throws Exception {

    Bucket bucket = mock(Bucket.class);
    String inFileDataStr = "doc-in";
    String content = "binary";
    ByteBuf buf = Unpooled.copiedBuffer(content.getBytes(StandardCharsets.UTF_8));
    when(bucket.get(inFileDataStr, BinaryDocument.class))
        .thenReturn(BinaryDocument.create(inFileDataStr, buf));
    setupMockBucket(bucket);

    byte[] inFileData = inFileDataStr.getBytes(StandardCharsets.UTF_8);
    testRunner.enqueue(inFileData);
    testRunner.setProperty(DOCUMENT_TYPE, DocumentType.Binary.toString());
    testRunner.setProperty(PUT_VALUE_TO_ATTRIBUTE, "targetAttribute");
    testRunner.run();

    testRunner.assertTransferCount(REL_SUCCESS, 1);
    testRunner.assertTransferCount(REL_ORIGINAL, 0);
    testRunner.assertTransferCount(REL_RETRY, 0);
    testRunner.assertTransferCount(REL_FAILURE, 0);
    MockFlowFile outFile = testRunner.getFlowFilesForRelationship(REL_SUCCESS).get(0);
    outFile.assertContentEquals(inFileDataStr);
    outFile.assertAttributeEquals("targetAttribute", "binary");
}
 

@Test
public void testGet() throws Exception {
    final CouchbaseMapCacheClient client = new CouchbaseMapCacheClient();
    final CouchbaseClusterControllerService couchbaseService = mock(CouchbaseClusterControllerService.class);
    final Bucket bucket = mock(Bucket.class);

    final MockControllerServiceInitializationContext serviceInitializationContext
            = new MockControllerServiceInitializationContext(couchbaseService, "couchbaseService");
    final Map<PropertyDescriptor, String> properties = new HashMap<>();
    properties.put(COUCHBASE_CLUSTER_SERVICE, "couchbaseService");
    properties.put(BUCKET_NAME, "bucketA");

    final ByteBuf contents = Unpooled.copiedBuffer("value".getBytes(StandardCharsets.UTF_8));
    final BinaryDocument doc = BinaryDocument.create("key", contents);
    when(couchbaseService.openBucket(eq("bucketA"))).thenReturn(bucket);
    when(bucket.get(any(BinaryDocument.class))).thenReturn(doc);

    final MockConfigurationContext context = new MockConfigurationContext(properties, serviceInitializationContext);
    client.configure(context);
    final String cacheEntry = client.get("key", stringSerializer, stringDeserializer);

    assertEquals("value", cacheEntry);
}
 

public EventIndexedRecord(ByteBuf value) {
    if (DcpMutationMessage.is(value)) {
        key = bufToString(DcpMutationMessage.key(value));
        seqno = DcpMutationMessage.bySeqno(value);
        event = "mutation";
        partition = DcpMutationMessage.partition(value);
        cas = DcpMutationMessage.cas(value);
        revSeqno = DcpMutationMessage.revisionSeqno(value);
        expiration = DcpMutationMessage.expiry(value);
        flags = DcpMutationMessage.flags(value);
        lockTime = DcpMutationMessage.lockTime(value);
        content = bufToBytes(DcpMutationMessage.content(value));
    } else if (DcpDeletionMessage.is(value)) {
        key = bufToString(DcpDeletionMessage.key(value));
        seqno = DcpDeletionMessage.bySeqno(value);
        event = "deletion";
        partition = DcpDeletionMessage.partition(value);
        cas = DcpDeletionMessage.cas(value);
        revSeqno = DcpDeletionMessage.revisionSeqno(value);
        expiration = 0;
        flags = 0;
        lockTime = 0;
        content = null;
    } else if (DcpExpirationMessage.is(value)) {
        key = bufToString(DcpExpirationMessage.key(value));
        seqno = DcpExpirationMessage.bySeqno(value);
        event = "expiration";
        partition = DcpExpirationMessage.partition(value);
        cas = DcpExpirationMessage.cas(value);
        revSeqno = DcpExpirationMessage.revisionSeqno(value);
        expiration = 0;
        flags = 0;
        lockTime = 0;
        content = null;
    } else {
        throw new IllegalArgumentException("Unexpected value type: " + value.getByte(1));
    }
}
 

public void startStreaming(final BlockingQueue<ByteBuf> resultsQueue) {
    if (streaming) {
        LOG.warn("This connection already in streaming mode, create another one.");
        return;
    }
    streaming = true;
    this.resultsQueue = resultsQueue;
    client.initializeState(StreamFrom.BEGINNING, StreamTo.NOW).await();
    new Thread(new Runnable() {
        @Override
        public void run() {
            int i = 0;
            try {
                client.startStreaming(partitionsToStream()).await();
                while (true) {
                    if (client.sessionState().isAtEnd()) {
                        break;
                    }
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        break;
                    }
                }
            } finally {
                streaming = false;
            }
        }
    }, "CouchbaseStreaming-" + threadId.incrementAndGet())
            .start();
}
 

public void stopStreaming() {
    if (resultsQueue != null) {
        client.stopStreaming(partitionsToStream()).await();
        BlockingQueue<ByteBuf> queue = resultsQueue;
        resultsQueue = null;
        List<ByteBuf> drained = new ArrayList<ByteBuf>();
        queue.drainTo(drained);
        for (ByteBuf byteBuf : drained) {
            byteBuf.release();
        }
        client.disconnect();
    }
}
 

@Test
public void testConvertToAvroMutation() {
    ByteBuf buffer = Mockito.mock(ByteBuf.class);

    // Mocking key object
    Mockito.when(buffer.getByte(0)).thenReturn(MessageUtil.MAGIC_REQ);
    Mockito.when(buffer.getByte(1)).thenReturn(MessageUtil.DCP_MUTATION_OPCODE);
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    ByteBuf key = Mockito.mock(ByteBuf.class);
    Mockito.when(key.readableBytes()).thenReturn(4);
    Mockito.when(buffer.slice(29, 10)).thenReturn(key);

    // Mocking content object.
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    Mockito.when(buffer.getInt(8)).thenReturn(100);
    ByteBuf content = Mockito.mock(ByteBuf.class);
    // Content byte array length.
    Mockito.when(content.readableBytes()).thenReturn(85);
    Mockito.when(buffer.slice(39, 85)).thenReturn(content);

    IndexedRecord record = converter.convertToAvro(buffer);

    Assert.assertEquals(schema, record.getSchema());

    assertIndexedRecordResult(record, "mutation", new byte[85]);

}
 

@Override
public Iterable<TupleDocument> convertRecord(String outputSchema, GenericRecord inputRecord, WorkUnitState workUnit)
    throws DataConversionException {
  String key = inputRecord.get(keyField).toString();
  GenericRecord data = (GenericRecord) inputRecord.get(dataRecordField);

  ByteBuffer dataBytes = (ByteBuffer) data.get(valueField);
  Integer flags = (Integer) data.get(flagsField);

  ByteBuf buffer = Unpooled.copiedBuffer(dataBytes);
  return new SingleRecordIterable<>(new TupleDocument(key, Tuple.create(buffer, flags)));
}
 

void onWrite(AbstractDocument doc)
    throws UnsupportedEncodingException {
  recordClass = doc.getClass();
  if (doc instanceof TupleDocument) {
    ByteBuf outgoingBuf = (((TupleDocument) doc).content()).value1();
    byte[] outgoingBytes = new byte[outgoingBuf.readableBytes()];
    outgoingBuf.getBytes(0, outgoingBytes);
    verificationCache.put(doc.id(), outgoingBytes);
  } else if (doc instanceof RawJsonDocument) {
    verificationCache.put(doc.id(), ((RawJsonDocument) doc).content().getBytes("UTF-8"));
  } else {
    throw new UnsupportedOperationException("Can only support TupleDocument or RawJsonDocument at this time");
  }
}
 

public static String getStringContent(Object content) {
    if (content instanceof String) {
        return (String) content;
    } else if (content instanceof byte[]) {
        return new String((byte[]) content, StandardCharsets.UTF_8);
    } else if (content instanceof ByteBuf) {
        final ByteBuf byteBuf = (ByteBuf) content;
        byte[] bytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(bytes);
        byteBuf.release();
        return new String(bytes, StandardCharsets.UTF_8);
    }
    return content.toString();
}
 

private <V> V deserialize(BinaryDocument doc, Deserializer<V> valueDeserializer) throws IOException {
    if (doc == null) {
        return null;
    }
    final ByteBuf byteBuf = doc.content();
    final byte[] bytes = new byte[byteBuf.readableBytes()];
    byteBuf.readBytes(bytes);
    byteBuf.release();
    return valueDeserializer.deserialize(bytes);
}
 

@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
    final ComponentLog logger = getLogger();
    FlowFile flowFile = session.get();
    if (flowFile == null) {
        return;
    }

    final byte[] content = new byte[(int) flowFile.getSize()];
    session.read(flowFile, new InputStreamCallback() {
        @Override
        public void process(final InputStream in) throws IOException {
            StreamUtils.fillBuffer(in, content, true);
        }
    });

    String docId = flowFile.getAttribute(CoreAttributes.UUID.key());
    if (!StringUtils.isEmpty(context.getProperty(DOC_ID).getValue())) {
        docId = context.getProperty(DOC_ID).evaluateAttributeExpressions(flowFile).getValue();
    }

    try {
        Document<?> doc = null;
        final DocumentType documentType = DocumentType.valueOf(context.getProperty(DOCUMENT_TYPE).getValue());
        switch (documentType) {
            case Json: {
                doc = RawJsonDocument.create(docId, new String(content, StandardCharsets.UTF_8));
                break;
            }
            case Binary: {
                final ByteBuf buf = Unpooled.copiedBuffer(content);
                doc = BinaryDocument.create(docId, buf);
                break;
            }
        }

        final PersistTo persistTo = PersistTo.valueOf(context.getProperty(PERSIST_TO).getValue());
        final ReplicateTo replicateTo = ReplicateTo.valueOf(context.getProperty(REPLICATE_TO).getValue());
        doc = openBucket(context).upsert(doc, persistTo, replicateTo);

        final Map<String, String> updatedAttrs = new HashMap<>();
        updatedAttrs.put(CouchbaseAttributes.Cluster.key(), context.getProperty(COUCHBASE_CLUSTER_SERVICE).getValue());
        updatedAttrs.put(CouchbaseAttributes.Bucket.key(), context.getProperty(BUCKET_NAME).getValue());
        updatedAttrs.put(CouchbaseAttributes.DocId.key(), docId);
        updatedAttrs.put(CouchbaseAttributes.Cas.key(), String.valueOf(doc.cas()));
        updatedAttrs.put(CouchbaseAttributes.Expiry.key(), String.valueOf(doc.expiry()));

        flowFile = session.putAllAttributes(flowFile, updatedAttrs);
        session.getProvenanceReporter().send(flowFile, getTransitUrl(context, docId));
        session.transfer(flowFile, REL_SUCCESS);
    } catch (final CouchbaseException e) {
        String errMsg = String.format("Writing document %s to Couchbase Server using %s failed due to %s", docId, flowFile, e);
        handleCouchbaseException(context, session, logger, flowFile, e, errMsg);
    }
}
 

@SuppressWarnings("unchecked")
private static CouchbaseConnector getConnector(Class responseClass) throws Exception {
  ENV = DefaultCouchbaseEnvironment.create();

  CouchbaseCore core = mock(CouchbaseCore.class);
  CouchbaseAsyncBucket asyncBucket = new CouchbaseAsyncBucket(core,
      ENV,
      BUCKET,
      USERNAME,
      PASSWORD,
      Collections.emptyList()
  );

  final CouchbaseRequest requestMock = mock(CouchbaseRequest.class);

  Subject<CouchbaseResponse, CouchbaseResponse> response = AsyncSubject.create();

  if(responseClass == SimpleSubdocResponse.class) {
    final BinarySubdocRequest subdocRequestMock = mock(BinarySubdocRequest.class);
    when(subdocRequestMock.span()).thenReturn(mock(Span.class));

    response.onNext(new SimpleSubdocResponse(ResponseStatus.SUCCESS,
        KeyValueStatus.SUCCESS.code(),
        BUCKET,
        Unpooled.EMPTY_BUFFER,
        subdocRequestMock,
        1234,
        null
    ));

    response.onCompleted();
  } else {

    Constructor con = responseClass.getConstructor(ResponseStatus.class,
        short.class,
        long.class,
        String.class,
        ByteBuf.class,
        MutationToken.class,
        CouchbaseRequest.class
    );

    response.onNext((CouchbaseResponse) con.newInstance(ResponseStatus.SUCCESS,
        KeyValueStatus.SUCCESS.code(),
        1234,
        BUCKET,
        Unpooled.EMPTY_BUFFER,
        null,
        requestMock
    ));
    response.onCompleted();
  }

  when(core.send(any(BinarySubdocRequest.class))).thenReturn(response);
  when(core.send(any())).thenReturn(response);
  when(requestMock.span()).thenReturn(mock(Span.class));

  CouchbaseConnector connector = mock(CouchbaseConnector.class);
  when(connector.getScheduler()).thenReturn(ENV.scheduler());
  when(connector.bucket()).thenReturn(asyncBucket);

  return connector;
}
 

private static byte[] bufToBytes(ByteBuf buf) {
    byte[] bytes;
    bytes = new byte[buf.readableBytes()];
    buf.readBytes(bytes);
    return bytes;
}
 

private static String bufToString(ByteBuf buf) {
    return new String(bufToBytes(buf), CharsetUtil.UTF_8);
}
 

@Override
public Class<ByteBuf> getDatumClass() {
    return ByteBuf.class;
}
 

@Override
public ByteBuf convertToDatum(IndexedRecord value) {
    throw new UnmodifiableAdapterException();
}
 

@Override
public IndexedRecord convertToAvro(ByteBuf value) {
    return new EventIndexedRecord(value);
}
 

public void acknowledge(ByteBuf event) {
    dataEventHandler.getController().ack(event);
}