Flink 1.4对接Elasticsearch 5系列连接器遇异常求助

Hey there, let's break down your problems step by step—since you're new to Flink, version compatibility and dependency conflicts are super common pitfalls here.

First: Clarify Version Compatibility #

Flink 1.4's official Elasticsearch 5.x connector is designed to work best with Elasticsearch 5.1.x versions, not 5.5. Elasticsearch makes subtle API changes even in minor versions, which is why your job got stuck in a waiting state when using ES 5.5—your Flink connector couldn't properly communicate with the ES cluster to initialize the connection.

Fixing the Jar Submission Error with ES 5.1.2 #

When you switched to ES 5.1.2 and hit an exception during Jar submission, here are the key things to check:

• Match Connector & Flink Versions Exactly: Make sure your Java project uses the Flink ES connector built for Flink 1.4. The correct Maven/Gradle coordinate is flink-connector-elasticsearch5_2.11:1.4.0—don't mix in newer ES client libraries or mismatched Flink dependencies.
• Avoid Fat Jar Conflicts: If you're building a fat jar that includes all dependencies, exclude Flink's core libraries (like flink-core, flink-runtime) because these are already provided by the Flink cluster. Using Maven's shade plugin with exclusion rules will prevent class duplication errors.
• Dig Into Flink Logs: Check the JobManager and TaskManager logs for the full exception stack trace. Common issues here include NoClassDefFoundError (from bad dependencies) or ConnectionRefused (if ES isn't accessible, or CORS isn't enabled—make sure http.cors.enabled is set to true in your ES config).

Advice on Upgrading Flink #

If you're considering upgrading Flink to resolve this, here's what you need to know:

• Target Flink 1.7+ for Better ES Support: Flink 1.7 improved compatibility with Elasticsearch 5.x and added support for ES 6.x. This would let you use ES 5.5 or newer without compatibility headaches.
• Adapt Your Code to New APIs: Flink changed some connector configuration APIs between 1.4 and 1.7, so you'll need to update your code to match the new setup patterns (like using ElasticsearchSink.Builder instead of older constructor-based configs).
• Double-Check Dependencies Post-Upgrade: For Flink 1.7, use flink-connector-elasticsearch5_2.11:1.7.2 for ES 5.x, or flink-connector-elasticsearch6_2.11:1.7.2 if you move to ES 6.x.

Quick Debugging Tips #

• Test ES Connection Outside Flink: Write a simple standalone Java program using the ES 5.1.2 client to connect to your cluster and run basic read/write operations. This will confirm if ES itself is working and accessible, ruling out network or ES config issues.
• Add Debug Logs to Your Job: Insert log statements right before initializing the Elasticsearch sink to verify your config values (like ES hosts, index names) are correct. A common mistake is including the http:// prefix in host addresses—Flink's connector expects just ip:9200.

内容的提问来源于stack exchange，提问作者user2007820