Skip to content

Log server version on sink task startup #561

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

Merged
merged 11 commits into from
Jun 28, 2021
Merged

Log server version on sink task startup #561

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
c4b0c02
Log server version on sink task startup. Revert client upgrade
Jun 28, 2021
b13ea6e
Fix spacing
Jun 28, 2021
fb85c05
Fix build by replacing Metadata packages
Jun 28, 2021
bd1fbd9
Catch all exceptions when getting ES server version. Close Client
Jun 28, 2021
8a332fc
set high level client to null on close
Jun 28, 2021
84c99bf
Unused Imports
Jun 28, 2021
1a660b1
Move comments. change log level to warn
Jun 28, 2021
4a129ea
Close high level client after use`
Jun 28, 2021
d6ac411
Wrap client close in try
Jun 28, 2021
94c8aa6
Change log level to warn
Jun 28, 2021
06d5112
Add error to log`
Jun 28, 2021
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
40 changes: 21 additions & 19 deletions src/main/java/io/confluent/connect/elasticsearch/ElasticsearchClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,25 +15,6 @@

package io.confluent.connect.elasticsearch;

import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.FLUSH_TIMEOUT_MS_CONFIG;
import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.MAX_BUFFERED_RECORDS_CONFIG;

import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BehaviorOnMalformedDoc;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import org.apache.http.HttpHost;
import org.apache.http.nio.conn.NHttpClientConnectionManager;
import org.apache.kafka.common.utils.Time;
Expand Down Expand Up @@ -62,6 +43,27 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BehaviorOnMalformedDoc;

import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.FLUSH_TIMEOUT_MS_CONFIG;
import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.MAX_BUFFERED_RECORDS_CONFIG;

public class ElasticsearchClient {

private static final Logger log = LoggerFactory.getLogger(ElasticsearchClient.class);
Expand Down
47 changes: 44 additions & 3 deletions src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@

package io.confluent.connect.elasticsearch;

import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BehaviorOnNullValues;
import org.apache.http.HttpHost;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.errors.ConnectException;
Expand All @@ -24,13 +24,20 @@
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.main.MainResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.BehaviorOnNullValues;

public class ElasticsearchSinkTask extends SinkTask {

Expand Down Expand Up @@ -62,7 +69,6 @@ protected void start(Map<String, String> props, ElasticsearchClient client) {
if (context.errantRecordReporter() == null) {
log.info("Errant record reporter not configured.");
}

// may be null if DLQ not enabled
reporter = context.errantRecordReporter();
} catch (NoClassDefFoundError | NoSuchMethodError e) {
Expand All @@ -72,7 +78,8 @@ protected void start(Map<String, String> props, ElasticsearchClient client) {

this.client = client != null ? client : new ElasticsearchClient(config, reporter);

log.info("Started ElasticsearchSinkTask.");
log.info("Started ElasticsearchSinkTask. Connecting to ES server version: {}",
Copy link
Member

@kpatelatwork kpatelatwork Jun 28, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in the future scapping logs wont work. We should file another enhancement ticket to log this version as a metric this way we can plot it in datadog

Copy link
Author

@ilanjiR ilanjiR Jun 28, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://confluentinc.atlassian.net/browse/CCMSG-1218

getServerVersion());
}

@Override
Expand Down Expand Up @@ -125,6 +132,40 @@ private void checkMapping(SinkRecord record) {
}
}

private String getServerVersion() {
ConfigCallbackHandler configCallbackHandler = new ConfigCallbackHandler(config);
RestHighLevelClient highLevelClient = new RestHighLevelClient(
RestClient
.builder(
config.connectionUrls()
.stream()
.map(HttpHost::create)
.collect(Collectors.toList())
.toArray(new HttpHost[config.connectionUrls().size()])
)
.setHttpClientConfigCallback(configCallbackHandler)
.setRequestConfigCallback(configCallbackHandler)
);
MainResponse response;
String esVersionNumber = "Unknown";
try {
response = highLevelClient.info(RequestOptions.DEFAULT);
esVersionNumber = response.getVersion().toString();
} catch (Exception e) {
// Same error messages as from validating the connection for IOException.
// Insufficient privileges to validate the version number if caught
// ElasticsearchStatusException.
log.warn("Failed to get ES server version", e);
} finally {
try {
highLevelClient.close();
} catch (Exception e) {
log.warn("Failed to close high level client", e);
}
}
return esVersionNumber;
}

/**
* Returns the converted index name from a given topic name. Elasticsearch accepts:
* <ul>
Expand Down
11 changes: 6 additions & 5 deletions src/test/java/io/confluent/connect/elasticsearch/helper/ElasticsearchHelperClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,6 @@

package io.confluent.connect.elasticsearch.helper;

import io.confluent.connect.elasticsearch.ConfigCallbackHandler;
import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig;
import java.io.IOException;
import java.util.Map.Entry;

import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.search.SearchRequest;
Expand All @@ -42,6 +37,12 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map.Entry;

import io.confluent.connect.elasticsearch.ConfigCallbackHandler;
import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig;

public class ElasticsearchHelperClient {

private static final Logger log = LoggerFactory.getLogger(ElasticsearchHelperClient.class);
Expand Down