Skip to content

Commit e54ec6e

Browse files
dlvenabledlvenable
authored
Initial addition of the PullIngester for the opensearch sink. (#6842)
Initial addition of the PullIngester for the opensearch sink. This creates a new PullIngester and implements the first one as the KafkaPullEngine. It reads an index to find the pull ingestion topic and then writes data to that topic. It routes shards using the same Murmur 3 approach that OpenSearch uses. The pull-based ingestion is marked as experimental. The configuration requires specifying the document Id currently. Resolves #6835 Signed-off-by: David Venable <dlv@amazon.com>
1 parent a18b56c commit e54ec6e

32 files changed

Lines changed: 2194 additions & 22 deletions

File tree

‎data-prepper-plugins/opensearch-pull-kafka/build.gradle‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
dependencies {
11+
implementation project(':data-prepper-api')
12+
implementation project(':data-prepper-plugins:opensearch')
13+
implementation 'org.apache.kafka:kafka-clients:3.9.1'
14+
implementation 'com.fasterxml.jackson.core:jackson-databind'
15+
implementation 'javax.inject:javax.inject:1'
16+
compileOnly 'org.projectlombok:lombok:1.18.46'
17+
annotationProcessor 'org.projectlombok:lombok:1.18.46'
18+
testImplementation project(':data-prepper-test:plugin-test-framework')
19+
}

‎data-prepper-plugins/opensearch-pull-kafka/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull/kafka/KafkaPullEngine.java‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.plugins.sink.opensearch.pull.kafka;
11+
12+
import org.apache.kafka.clients.producer.KafkaProducer;
13+
import org.apache.kafka.clients.producer.ProducerConfig;
14+
import org.apache.kafka.clients.producer.ProducerRecord;
15+
import org.apache.kafka.common.serialization.ByteArraySerializer;
16+
import org.apache.kafka.common.serialization.StringSerializer;
17+
import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
18+
import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
19+
import org.opensearch.dataprepper.plugins.sink.opensearch.PullEngine;
20+
import org.slf4j.Logger;
21+
import org.slf4j.LoggerFactory;
22+
23+
import java.util.Properties;
24+
25+
@DataPrepperPlugin(name = "kafka", pluginType = PullEngine.class, pluginConfigurationType = KafkaPullEngineConfig.class,
26+
packagesToScan = {KafkaPullEngine.class})
27+
public class KafkaPullEngine implements PullEngine {
28+
private static final Logger LOG = LoggerFactory.getLogger(KafkaPullEngine.class);
29+
30+
private final String bootstrapServers;
31+
private final TopicManager topicManager;
32+
33+
private KafkaProducer<String, byte[]> producer;
34+
private String topicName;
35+
36+
@DataPrepperPluginConstructor
37+
public KafkaPullEngine(final KafkaPullEngineConfig config, final TopicManager topicManager) {
38+
this.bootstrapServers = String.join(",", config.getBootstrapServers());
39+
this.topicManager = topicManager;
40+
}
41+
42+
@Override
43+
public void initialize(final String topicName, final int partitionCount) {
44+
this.topicName = topicName;
45+
46+
topicManager.createTopicWithPartitions(topicName, partitionCount);
47+
48+
final Properties producerProps = new Properties();
49+
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
50+
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
51+
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
52+
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
53+
54+
this.producer = new KafkaProducer<>(producerProps);
55+
LOG.info("KafkaPullEngine initialized with topic '{}' and {} partition(s)", topicName, partitionCount);
56+
}
57+
58+
@Override
59+
public void write(final int partition, final String key, final byte[] document) {
60+
final ProducerRecord<String, byte[]> record = new ProducerRecord<>(topicName, partition, key, document);
61+
producer.send(record, (metadata, exception) -> {
62+
if (exception != null) {
63+
LOG.error("Failed to send record with key '{}' to partition {}", key, partition, exception);
64+
}
65+
});
66+
}
67+
68+
@Override
69+
public void flush() {
70+
if (producer != null) {
71+
producer.flush();
72+
}
73+
}
74+
75+
@Override
76+
public void shutdown() {
77+
if (producer != null) {
78+
producer.flush();
79+
producer.close();
80+
}
81+
}
82+
}

‎data-prepper-plugins/opensearch-pull-kafka/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull/kafka/KafkaPullEngineConfig.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.plugins.sink.opensearch.pull.kafka;
11+
12+
import com.fasterxml.jackson.annotation.JsonProperty;
13+
import jakarta.validation.constraints.NotEmpty;
14+
import jakarta.validation.constraints.NotNull;
15+
import lombok.Getter;
16+
17+
import java.util.List;
18+
19+
public class KafkaPullEngineConfig {
20+
@Getter
21+
@JsonProperty("bootstrap_servers")
22+
@NotNull
23+
@NotEmpty
24+
private List<String> bootstrapServers;
25+
}

‎data-prepper-plugins/opensearch-pull-kafka/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull/kafka/TopicManager.java‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.plugins.sink.opensearch.pull.kafka;
11+
12+
import org.apache.kafka.clients.admin.AdminClient;
13+
import org.apache.kafka.clients.admin.AdminClientConfig;
14+
import org.apache.kafka.clients.admin.NewPartitions;
15+
import org.apache.kafka.clients.admin.NewTopic;
16+
import org.apache.kafka.clients.admin.TopicDescription;
17+
import org.slf4j.Logger;
18+
import org.slf4j.LoggerFactory;
19+
20+
import javax.inject.Named;
21+
import java.util.Collections;
22+
import java.util.Map;
23+
import java.util.Properties;
24+
import java.util.concurrent.ExecutionException;
25+
import java.util.concurrent.TimeUnit;
26+
import java.util.concurrent.TimeoutException;
27+
28+
import org.apache.kafka.common.errors.InvalidPartitionsException;
29+
import org.apache.kafka.common.errors.ReassignmentInProgressException;
30+
31+
@Named
32+
public class TopicManager {
33+
private static final Logger LOG = LoggerFactory.getLogger(TopicManager.class);
34+
private static final long TOPIC_READY_TIMEOUT_MS = 30_000;
35+
private static final long PARTITION_POLL_INTERVAL_MS = 500;
36+
private static final int MAX_PARTITION_RETRIES = 3;
37+
private static final long PARTITION_RETRY_BACKOFF_MS = 1_000;
38+
39+
private final String bootstrapServers;
40+
41+
public TopicManager(final KafkaPullEngineConfig config) {
42+
this.bootstrapServers = String.join(",", config.getBootstrapServers());
43+
}
44+
45+
void createTopicWithPartitions(final String topicName, final int partitionCount) {
46+
final Properties adminProps = new Properties();
47+
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
48+
49+
try (final AdminClient adminClient = AdminClient.create(adminProps)) {
50+
final NewTopic newTopic = new NewTopic(topicName, partitionCount, (short) 1);
51+
adminClient.createTopics(Collections.singleton(newTopic)).all().get();
52+
LOG.info("Created Kafka topic '{}' with {} partition(s)", topicName, partitionCount);
53+
waitForPartitions(adminClient, topicName, partitionCount);
54+
} catch (final ExecutionException e) {
55+
if (e.getCause() instanceof org.apache.kafka.common.errors.TopicExistsException) {
56+
LOG.info("Kafka topic '{}' already exists, verifying partition count", topicName);
57+
ensurePartitionCount(topicName, partitionCount);
58+
} else {
59+
throw new RuntimeException("Failed to create Kafka topic: " + topicName, e);
60+
}
61+
} catch (final InterruptedException e) {
62+
Thread.currentThread().interrupt();
63+
throw new RuntimeException("Interrupted while creating Kafka topic: " + topicName, e);
64+
}
65+
}
66+
67+
private void ensurePartitionCount(final String topicName, final int requiredPartitions) {
68+
final Properties adminProps = new Properties();
69+
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
70+
71+
try (final AdminClient adminClient = AdminClient.create(adminProps)) {
72+
final Map<String, TopicDescription> descriptions = adminClient
73+
.describeTopics(Collections.singleton(topicName))
74+
.allTopicNames()
75+
.get(TOPIC_READY_TIMEOUT_MS, TimeUnit.MILLISECONDS);
76+
final TopicDescription description = descriptions.get(topicName);
77+
final int currentPartitions = description.partitions().size();
78+
79+
if (currentPartitions < requiredPartitions) {
80+
LOG.info("Topic '{}' has {} partition(s) but {} required, increasing partition count",
81+
topicName, currentPartitions, requiredPartitions);
82+
increasePartitionsWithRetry(adminClient, topicName, requiredPartitions);
83+
waitForPartitions(adminClient, topicName, requiredPartitions);
84+
} else {
85+
LOG.info("Topic '{}' already has {} partition(s), required {}", topicName, currentPartitions, requiredPartitions);
86+
}
87+
} catch (final ExecutionException | TimeoutException e) {
88+
throw new RuntimeException("Failed to verify/update partition count for topic: " + topicName, e);
89+
} catch (final InterruptedException e) {
90+
Thread.currentThread().interrupt();
91+
throw new RuntimeException("Interrupted while verifying partition count for topic: " + topicName, e);
92+
}
93+
}
94+
95+
private void increasePartitionsWithRetry(final AdminClient adminClient, final String topicName, final int requiredPartitions) throws InterruptedException {
96+
for (int attempt = 1; attempt <= MAX_PARTITION_RETRIES; attempt++) {
97+
try {
98+
adminClient.createPartitions(
99+
Collections.singletonMap(topicName, NewPartitions.increaseTo(requiredPartitions))
100+
).all().get();
101+
LOG.info("Increased partition count for topic '{}' to {}", topicName, requiredPartitions);
102+
return;
103+
} catch (final ExecutionException e) {
104+
final Throwable cause = e.getCause();
105+
if ((cause instanceof InvalidPartitionsException || cause instanceof ReassignmentInProgressException)
106+
&& attempt < MAX_PARTITION_RETRIES) {
107+
LOG.warn("Partition reassignment conflict on topic '{}' (attempt {}/{}), retrying after backoff",
108+
topicName, attempt, MAX_PARTITION_RETRIES, cause);
109+
Thread.sleep(PARTITION_RETRY_BACKOFF_MS * attempt);
110+
} else {
111+
throw new RuntimeException("Failed to increase partition count for topic: " + topicName, e);
112+
}
113+
}
114+
}
115+
}
116+
117+
private void waitForPartitions(final AdminClient adminClient, final String topicName, final int expectedPartitions) {
118+
final long deadline = System.currentTimeMillis() + TOPIC_READY_TIMEOUT_MS;
119+
while (System.currentTimeMillis() < deadline) {
120+
try {
121+
final Map<String, TopicDescription> descriptions = adminClient
122+
.describeTopics(Collections.singleton(topicName))
123+
.allTopicNames()
124+
.get(TOPIC_READY_TIMEOUT_MS, TimeUnit.MILLISECONDS);
125+
final TopicDescription description = descriptions.get(topicName);
126+
if (description != null && description.partitions().size() >= expectedPartitions) {
127+
LOG.info("All {} partition(s) ready for topic '{}'", expectedPartitions, topicName);
128+
return;
129+
}
130+
} catch (final InterruptedException e) {
131+
Thread.currentThread().interrupt();
132+
throw new RuntimeException("Interrupted while waiting for topic partitions", e);
133+
} catch (final ExecutionException | TimeoutException e) {
134+
LOG.debug("Waiting for topic '{}' partitions to become available", topicName);
135+
}
136+
try {
137+
Thread.sleep(PARTITION_POLL_INTERVAL_MS);
138+
} catch (final InterruptedException e) {
139+
Thread.currentThread().interrupt();
140+
throw new RuntimeException("Interrupted while waiting for topic partitions", e);
141+
}
142+
}
143+
LOG.warn("Timed out waiting for all {} partition(s) on topic '{}' to become available", expectedPartitions, topicName);
144+
}
145+
}

‎data-prepper-plugins/opensearch-pull-kafka/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/pull/kafka/KafkaPluginEngineIT.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
/*
2+
* Copyright OpenSearch Contributors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*
5+
* The OpenSearch Contributors require contributions made to
6+
* this file be licensed under the Apache-2.0 license or a
7+
* compatible open source license.
8+
*/
9+
10+
package org.opensearch.dataprepper.plugins.sink.opensearch.pull.kafka;
11+
12+
import org.junit.jupiter.api.Test;
13+
import org.opensearch.dataprepper.plugins.sink.opensearch.PullEngine;
14+
import org.opensearch.dataprepper.test.plugins.DataPrepperPluginTest;
15+
import org.opensearch.dataprepper.test.plugins.PluginConfigurationFile;
16+
import org.opensearch.dataprepper.test.plugins.junit.BaseDataPrepperPluginStandardTestSuite;
17+
18+
import static org.hamcrest.CoreMatchers.instanceOf;
19+
import static org.hamcrest.CoreMatchers.notNullValue;
20+
import static org.hamcrest.MatcherAssert.assertThat;
21+
22+
@DataPrepperPluginTest(pluginName = "kafka", pluginType = PullEngine.class)
23+
public class KafkaPluginEngineIT extends BaseDataPrepperPluginStandardTestSuite {
24+
@Test
25+
void constructs_with_valid_configuration(
26+
@PluginConfigurationFile("valid_kafka_config.yaml") final PullEngine pullEngine) {
27+
assertThat(pullEngine, notNullValue());
28+
assertThat(pullEngine, instanceOf(KafkaPullEngine.class));
29+
}
30+
}

0 commit comments

Comments
 (0)