Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[FEATURE] Add Support for querying Avro Data in Kafka Connector #1958

Merged
merged 3 commits into from
Jun 13, 2024
Merged

[FEATURE] Add Support for querying Avro Data in Kafka Connector #1958

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b694bc2
Add Support for querying Avro Data in Kafka Connector
Jithendar12 May 3, 2024
715a732
Apply Glue Schema Name to Kafka Topic Name
Jithendar12 May 8, 2024
93eb8ff
Close Kafka Consumer using try-with-resources statement
Jithendar12 Jun 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Close Kafka Consumer using try-with-resources statement
  • Loading branch information
@Jithendar12 @AbdulR3hman
Jithendar12 authored and AbdulR3hman committed Jun 13, 2024
commit 93eb8ff23a2b90b1e46770fcfab708296786ff51
14 changes: 4 additions & 10 deletions athena-kafka/src/main/java/com/amazonaws/athena/connectors/kafka/KafkaRecordHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,7 @@ private void consume(
{
LOGGER.info("[kafka] {} Polling for data", splitParameters);
int emptyResultFoundCount = 0;
try {
try (Consumer<String, TopicResultSet> consumer = kafkaConsumer) {
while (true) {
if (!queryStatusChecker.isQueryRunning()) {
LOGGER.debug("[kafka]{} Stopping and closing consumer due to query execution terminated by athena", splitParameters);
Expand All @@ -175,7 +175,7 @@ private void consume(

// Call the poll on consumer to fetch data from kafka server
// poll returns data as batch which can be configured.
ConsumerRecords<String, TopicResultSet> records = kafkaConsumer.poll(Duration.ofSeconds(1L));
ConsumerRecords<String, TopicResultSet> records = consumer.poll(Duration.ofSeconds(1L));
LOGGER.debug("[kafka] {} polled records size {}", splitParameters, records.count());

// For debug insight
Expand Down Expand Up @@ -216,9 +216,6 @@ private void consume(
}
}
}
finally {
kafkaConsumer.close();
}
}

/**
Expand Down Expand Up @@ -261,7 +258,7 @@ private void avroConsume(
{
LOGGER.info("[kafka] {} Polling for data", splitParameters);
int emptyResultFoundCount = 0;
try {
try (Consumer<String, GenericRecord> avroConsumer = kafkaAvroConsumer) {
while (true) {
if (!queryStatusChecker.isQueryRunning()) {
LOGGER.debug("[kafka]{} Stopping and closing consumer due to query execution terminated by athena", splitParameters);
Expand All @@ -271,7 +268,7 @@ private void avroConsume(

// Call the poll on consumer to fetch data from kafka server
// poll returns data as batch which can be configured.
ConsumerRecords<String, GenericRecord> records = kafkaAvroConsumer.poll(Duration.ofSeconds(1L));
ConsumerRecords<String, GenericRecord> records = avroConsumer.poll(Duration.ofSeconds(1L));
LOGGER.debug("[kafka] {} polled records size {}", splitParameters, records.count());

// For debug insight
Expand Down Expand Up @@ -312,9 +309,6 @@ private void avroConsume(
}
}
}
finally {
kafkaAvroConsumer.close();
}
}

private void avroExecute(
Expand Down