<commit_before>/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.server.eventhubs;

import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.enterprise.context.Dependent;
import javax.enterprise.inject.Instance;
import javax.inject.Inject;
import javax.inject.Named;

import org.eclipse.microprofile.config.Config;
import org.eclipse.microprofile.config.ConfigProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.azure.core.amqp.exception.AmqpException;
import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventDataBatch;
import com.azure.messaging.eventhubs.EventHubClientBuilder;
import com.azure.messaging.eventhubs.EventHubProducerClient;
import com.azure.messaging.eventhubs.models.CreateBatchOptions;

import io.debezium.DebeziumException;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.DebeziumEngine.RecordCommitter;
import io.debezium.server.BaseChangeConsumer;
import io.debezium.server.CustomConsumerBuilder;

/**
 * This sink adapter delivers change event messages to Azure Event Hubs
 *
 * @author <NAME>
 *
 */
@Named("eventhubs")
@Dependent
public class EventHubsChangeConsumer extends BaseChangeConsumer
        implements DebeziumEngine.ChangeConsumer<ChangeEvent<Object, Object>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubsChangeConsumer.class);

    private static final String PROP_PREFIX = "debezium.sink.eventhubs.";
    private static final String PROP_CONNECTION_STRING_NAME = PROP_PREFIX + "connectionstring";
    private static final String PROP_EVENTHUB_NAME = PROP_PREFIX + "hubname";
    private static final String PROP_PARTITION_ID = PROP_PREFIX + "partitionid";
    private static final String PROP_PARTITION_KEY = PROP_PREFIX + "partitionkey";
    // maximum size for the batch of events (bytes)
    private static final String PROP_MAX_BATCH_SIZE = PROP_PREFIX + "maxbatchsize";

    private String connectionString;
    private String eventHubName;
    private String partitionID;
    private String partitionKey;
    private Integer maxBatchSize;

    // connection string format -
    // Endpoint=sb://<NAMESPACE>/;SharedAccessKeyName=<KEY_NAME>;SharedAccessKey=<ACCESS_KEY>;EntityPath=<HUB_NAME>
    private static final String CONNECTION_STRING_FORMAT = "%s;EntityPath=%s";

    private EventHubProducerClient producer = null;

    @Inject
    @CustomConsumerBuilder
    Instance<EventHubProducerClient> customProducer;

    @PostConstruct
    void connect() {
        if (customProducer.isResolvable()) {
            producer = customProducer.get();
            LOGGER.info("Obtained custom configured Event Hubs client for namespace '{}'",
                    customProducer.get().getFullyQualifiedNamespace());
            return;
        }

        final Config config = ConfigProvider.getConfig();
        connectionString = config.getValue(PROP_CONNECTION_STRING_NAME, String.class);
        eventHubName = config.getValue(PROP_EVENTHUB_NAME, String.class);

        // optional config
        partitionID = config.getOptionalValue(PROP_PARTITION_ID, String.class).orElse("");
        partitionKey = config.getOptionalValue(PROP_PARTITION_KEY, String.class).orElse("");
        maxBatchSize = config.getOptionalValue(PROP_MAX_BATCH_SIZE, Integer.class).orElse(0);

        String finalConnectionString = String.format(CONNECTION_STRING_FORMAT, connectionString, eventHubName);

        try {
            producer = new EventHubClientBuilder().connectionString(finalConnectionString).buildProducerClient();
        }
        catch (Exception e) {
            throw new DebeziumException(e);
        }

        LOGGER.info("Using default Event Hubs client for namespace '{}'", producer.getFullyQualifiedNamespace());
    }

    @PreDestroy
    void close() {
        try {
            producer.close();
            LOGGER.info("Closed Event Hubs producer client");
        }
        catch (Exception e) {
            LOGGER.warn("Exception while closing Event Hubs producer: {}", e);
        }
    }

    @Override
    public void handleBatch(List<ChangeEvent<Object, Object>> records,
                            RecordCommitter<ChangeEvent<Object, Object>> committer)
            throws InterruptedException {
        LOGGER.trace("Processing change events...");

        CreateBatchOptions op = new CreateBatchOptions().setPartitionId(partitionID);
        if (partitionKey != "") {
            op.setPartitionKey(partitionKey);
        }
        if (maxBatchSize.intValue() != 0) {
            op.setMaximumSizeInBytes(maxBatchSize);
        }

        EventDataBatch batch = producer.createBatch(op);

        for (ChangeEvent<Object, Object> record : records) {
            LOGGER.trace("Received event '{}'", record);
            if (null == record.value()) {
                continue;
            }
            EventData eventData = null;

            if (record.value() instanceof String) {
                eventData = new EventData((String) record.value());
            }
            else if (record.value() instanceof byte[]) {
                eventData = new EventData(getBytes(record.value()));
            }
            try {
                if (!batch.tryAdd(eventData)) {
                    LOGGER.warn("event was too large to fit in the batch - {}", record);
                    continue;
                }
            }
            catch (IllegalArgumentException e) {
                LOGGER.warn("EventData was null - {}", e.getMessage());
                continue;
            }
            catch (AmqpException e) {
                LOGGER.warn("EventData is larger than the maximum size of the EventDataBatch - {}", e.getMessage());
                continue;
            }
            catch (Exception e) {
                LOGGER.warn("Failed to add EventData to batch {}", e.getMessage());
                continue;
            }

            // Event Hubs producer only supports "batch"ed sends. Each record is sent as a
            // separate batch which is then acknowledged/committed

            try {
                producer.send(batch);
                LOGGER.trace("Sent record to Event Hubs");
            }
            catch (Exception e) {
                LOGGER.warn("Failed to send record to Event Hubs {}", e.getMessage());
                // do not mark the record as processed it its not sent to Event Hubs
                continue;
            }
            try {
                committer.markProcessed(record);
                LOGGER.trace("Record marked processed");
            }
            catch (Exception e) {
                LOGGER.warn("Failed to mark record as processed {}", e.getMessage());
            }
        }

        committer.markBatchFinished();
        LOGGER.info("Batch marked finished");
    }
}
<commit_msg>DBZ-2282 updated to use batching<commit_after>/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.server.eventhubs;

import java.util.List;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.enterprise.context.Dependent;
import javax.enterprise.inject.Instance;
import javax.inject.Inject;
import javax.inject.Named;

import org.eclipse.microprofile.config.Config;
import org.eclipse.microprofile.config.ConfigProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.azure.core.amqp.exception.AmqpException;
import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventDataBatch;
import com.azure.messaging.eventhubs.EventHubClientBuilder;
import com.azure.messaging.eventhubs.EventHubProducerClient;
import com.azure.messaging.eventhubs.models.CreateBatchOptions;

import io.debezium.DebeziumException;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.DebeziumEngine.RecordCommitter;
import io.debezium.server.BaseChangeConsumer;
import io.debezium.server.CustomConsumerBuilder;

/**
 * This sink adapter delivers change event messages to Azure Event Hubs
 *
 * @author <NAME>
 *
 */
@Named("eventhubs")
@Dependent
public class EventHubsChangeConsumer extends BaseChangeConsumer
        implements DebeziumEngine.ChangeConsumer<ChangeEvent<Object, Object>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubsChangeConsumer.class);

    private static final String PROP_PREFIX = "debezium.sink.eventhubs.";
    private static final String PROP_CONNECTION_STRING_NAME = PROP_PREFIX + "connectionstring";
    private static final String PROP_EVENTHUB_NAME = PROP_PREFIX + "hubname";
    private static final String PROP_PARTITION_ID = PROP_PREFIX + "partitionid";
    private static final String PROP_PARTITION_KEY = PROP_PREFIX + "partitionkey";
    // maximum size for the batch of events (bytes)
    private static final String PROP_MAX_BATCH_SIZE = PROP_PREFIX + "maxbatchsize";

    private String connectionString;
    private String eventHubName;
    private String partitionID;
    private String partitionKey;
    private Integer maxBatchSize;

    // connection string format -
    // Endpoint=sb://<NAMESPACE>/;SharedAccessKeyName=<KEY_NAME>;SharedAccessKey=<ACCESS_KEY>;EntityPath=<HUB_NAME>
    private static final String CONNECTION_STRING_FORMAT = "%s;EntityPath=%s";

    private EventHubProducerClient producer = null;

    @Inject
    @CustomConsumerBuilder
    Instance<EventHubProducerClient> customProducer;

    @PostConstruct
    void connect() {
        if (customProducer.isResolvable()) {
            producer = customProducer.get();
            LOGGER.info("Obtained custom configured Event Hubs client for namespace '{}'",
                    customProducer.get().getFullyQualifiedNamespace());
            return;
        }

        final Config config = ConfigProvider.getConfig();
        connectionString = config.getValue(PROP_CONNECTION_STRING_NAME, String.class);
        eventHubName = config.getValue(PROP_EVENTHUB_NAME, String.class);

        // optional config
        partitionID = config.getOptionalValue(PROP_PARTITION_ID, String.class).orElse("");
        partitionKey = config.getOptionalValue(PROP_PARTITION_KEY, String.class).orElse("");
        maxBatchSize = config.getOptionalValue(PROP_MAX_BATCH_SIZE, Integer.class).orElse(0);

        String finalConnectionString = String.format(CONNECTION_STRING_FORMAT, connectionString, eventHubName);

        try {
            producer = new EventHubClientBuilder().connectionString(finalConnectionString).buildProducerClient();
        } catch (Exception e) {
            throw new DebeziumException(e);
        }

        LOGGER.info("Using default Event Hubs client for namespace '{}'", producer.getFullyQualifiedNamespace());
    }

    @PreDestroy
    void close() {
        try {
            producer.close();
            LOGGER.info("Closed Event Hubs producer client");
        } catch (Exception e) {
            LOGGER.warn("Exception while closing Event Hubs producer: {}", e);
        }
    }

    @Override
    public void handleBatch(List<ChangeEvent<Object, Object>> records,
            RecordCommitter<ChangeEvent<Object, Object>> committer) throws InterruptedException {
        LOGGER.trace("Event Hubs sink adapter processing change events");

        CreateBatchOptions op = new CreateBatchOptions().setPartitionId(partitionID);
        if (partitionKey != "") {
            op.setPartitionKey(partitionKey);
        }
        if (maxBatchSize.intValue() != 0) {
            op.setMaximumSizeInBytes(maxBatchSize);
        }

        EventDataBatch batch = producer.createBatch(op);

        // this loop just adds records to the batch
        for (ChangeEvent<Object, Object> record : records) {
            LOGGER.trace("Received event '{}'", record);
            if (null == record.value()) {
                continue;
            }
            EventData eventData = null;

            if (record.value() instanceof String) {
                eventData = new EventData((String) record.value());
            } else if (record.value() instanceof byte[]) {
                eventData = new EventData(getBytes(record.value()));
            }
            try {
                if (!batch.tryAdd(eventData)) {
                    LOGGER.warn("Event data was too large to fit in the batch - {}", record);
                }
            } catch (IllegalArgumentException e) {
                LOGGER.warn("Event data was null - {}", e.getMessage());
            } catch (AmqpException e) {
                LOGGER.warn("Event data is larger than the maximum size of the EventDataBatch - {}", e.getMessage());
            } catch (Exception e) {
                LOGGER.warn("Failed to add event data to batch - {}", e.getMessage());
            }
        }

        try {
            producer.send(batch);
            LOGGER.trace("Sent record batch to Event Hubs");
        } catch (Exception e) {
            LOGGER.warn("Failed to send record to Event Hubs {}", e.getMessage());
        }

        // this loop commits each record
        for (ChangeEvent<Object, Object> record : records) {
            try {
                committer.markProcessed(record);
                LOGGER.trace("Record marked processed");
            } catch (Exception e) {
                LOGGER.warn("Failed to mark record as processed {}", e.getMessage());
            }
        }

        committer.markBatchFinished();
        LOGGER.trace("Batch marked finished");
    }
}

{'fraction_non_alphanumeric': 0.09703113685734975, 'fraction_numerical': 0.001448225923244026, 'mean_word_length': 2.755029907558456, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6278396', 'n_tokens_mistral': 3669, 'n_tokens_neox': 3642, 'n_words': 859}