|
37 | 37 | import org.apache.kafka.connect.storage.Converter; |
38 | 38 | import org.apache.kafka.connect.storage.HeaderConverter; |
39 | 39 | import org.apache.kafka.connect.storage.SimpleHeaderConverter; |
40 | | -import org.elasticsearch.action.DocWriteRequest; |
41 | | -import org.elasticsearch.action.DocWriteRequest.OpType; |
42 | | -import org.elasticsearch.action.delete.DeleteRequest; |
43 | | -import org.elasticsearch.action.index.IndexRequest; |
44 | | -import org.elasticsearch.action.update.UpdateRequest; |
45 | | -import org.elasticsearch.index.VersionType; |
46 | | -import org.elasticsearch.xcontent.XContentType; |
| 40 | +import co.elastic.clients.elasticsearch._types.VersionType; |
| 41 | +import co.elastic.clients.elasticsearch.core.bulk.BulkOperation; |
| 42 | +import co.elastic.clients.json.JsonData; |
| 43 | + |
47 | 44 | import org.slf4j.Logger; |
48 | 45 | import org.slf4j.LoggerFactory; |
49 | 46 |
|
@@ -121,7 +118,7 @@ private String convertKey(Schema keySchema, Object key) { |
121 | 118 | } |
122 | 119 | } |
123 | 120 |
|
124 | | - public DocWriteRequest<?> convertRecord(SinkRecord record, String resourceName) { |
| 121 | + public BulkOperation convertRecord(SinkRecord record, String resourceName) { |
125 | 122 | if (record.value() == null) { |
126 | 123 | switch (config.behaviorOnNullValues()) { |
127 | 124 | case IGNORE: |
@@ -166,32 +163,91 @@ public DocWriteRequest<?> convertRecord(SinkRecord record, String resourceName) |
166 | 163 |
|
167 | 164 | // delete |
168 | 165 | if (record.value() == null) { |
169 | | - return maybeAddExternalVersioning(new DeleteRequest(resourceName).id(id), record); |
| 166 | + if (!config.isDataStream() && !config.shouldIgnoreKey(record.topic())) { |
| 167 | + final long externalVersion = resolveExternalVersion(record); |
| 168 | + return BulkOperation.of(b -> b.delete(d -> d |
| 169 | + .index(resourceName).id(id) |
| 170 | + .versionType(VersionType.External).version(externalVersion))); |
| 171 | + } |
| 172 | + return BulkOperation.of(b -> b.delete(d -> d.index(resourceName).id(id))); |
170 | 173 | } |
171 | 174 |
|
172 | 175 | String payload = getPayload(record); |
173 | 176 | payload = maybeAddTimestamp(payload, record.timestamp()); |
| 177 | + final String finalPayload = payload; |
174 | 178 |
|
175 | 179 | // index |
176 | 180 | switch (config.writeMethod()) { |
177 | 181 | case UPSERT: |
178 | | - return new UpdateRequest(resourceName, id) |
179 | | - .doc(payload, XContentType.JSON) |
180 | | - .upsert(payload, XContentType.JSON) |
181 | | - .retryOnConflict(Math.min(config.maxInFlightRequests(), 5)); |
| 182 | + return BulkOperation.of(b -> b.update(u -> u |
| 183 | + .index(resourceName) |
| 184 | + .id(id) |
| 185 | + .retryOnConflict(Math.min(config.maxInFlightRequests(), 5)) |
| 186 | + .action(a -> a |
| 187 | + .doc(JsonData.fromJson(finalPayload)) |
| 188 | + .upsert(JsonData.fromJson(finalPayload)) |
| 189 | + ) |
| 190 | + )); |
182 | 191 | case INSERT: |
183 | | - OpType opType = config.isDataStream() ? OpType.CREATE : OpType.INDEX; |
184 | | - IndexRequest req = |
185 | | - new IndexRequest(resourceName).source(payload, XContentType.JSON).opType(opType); |
| 192 | + if (config.isDataStream()) { |
| 193 | + if (config.useAutogeneratedIds()) { |
| 194 | + return BulkOperation.of(b -> b.create(c -> c |
| 195 | + .index(resourceName) |
| 196 | + .document(JsonData.fromJson(finalPayload)) |
| 197 | + )); |
| 198 | + } |
| 199 | + return BulkOperation.of(b -> b.create(c -> c |
| 200 | + .index(resourceName).id(id) |
| 201 | + .document(JsonData.fromJson(finalPayload)) |
| 202 | + )); |
| 203 | + } |
186 | 204 | if (config.useAutogeneratedIds()) { |
187 | | - return req; |
| 205 | + return BulkOperation.of(b -> b.index(i -> i |
| 206 | + .index(resourceName) |
| 207 | + .document(JsonData.fromJson(finalPayload)) |
| 208 | + )); |
188 | 209 | } |
189 | | - return maybeAddExternalVersioning(req.id(id), record); |
| 210 | + if (!config.shouldIgnoreKey(record.topic())) { |
| 211 | + final long externalVersion = resolveExternalVersion(record); |
| 212 | + return BulkOperation.of(b -> b.index(i -> i |
| 213 | + .index(resourceName).id(id) |
| 214 | + .document(JsonData.fromJson(finalPayload)) |
| 215 | + .versionType(VersionType.External).version(externalVersion))); |
| 216 | + } |
| 217 | + return BulkOperation.of(b -> b.index(i -> i |
| 218 | + .index(resourceName).id(id) |
| 219 | + .document(JsonData.fromJson(finalPayload)) |
| 220 | + )); |
190 | 221 | default: |
191 | 222 | return null; // shouldn't happen |
192 | 223 | } |
193 | 224 | } |
194 | 225 |
|
| 226 | + /** |
| 227 | + * In many cases, we explicitly set the record version using the topic's offset. |
| 228 | + * This version will, in turn, be checked by Elasticsearch and will throw a versioning |
| 229 | + * error if the request represents an equivalent or older version of the record. |
| 230 | + * |
| 231 | + * @param record the record to be processed |
| 232 | + * @return the version to use for the Elasticsearch request |
| 233 | + */ |
| 234 | + private long resolveExternalVersion(SinkRecord record) { |
| 235 | + if (config.hasExternalVersionHeader()) { |
| 236 | + final Header versionHeader = record.headers().lastWithName(config.externalVersionHeader()); |
| 237 | + final byte[] versionValue = HEADER_CONVERTER.fromConnectHeader( |
| 238 | + record.topic(), versionHeader.key(), versionHeader.schema(), versionHeader.value() |
| 239 | + ); |
| 240 | + try { |
| 241 | + //fromConnectHeader byte output is UTF_8 |
| 242 | + return Long.parseLong(new String(versionValue, StandardCharsets.UTF_8)); |
| 243 | + } catch (NumberFormatException e) { |
| 244 | + throw new ConnectException("Error converting to long: " |
| 245 | + + new String(versionValue, StandardCharsets.UTF_8), e); |
| 246 | + } |
| 247 | + } |
| 248 | + return record.kafkaOffset(); |
| 249 | + } |
| 250 | + |
195 | 251 | private String getPayload(SinkRecord record) { |
196 | 252 | if (record.value() == null) { |
197 | 253 | return null; |
@@ -241,44 +297,6 @@ private String maybeAddTimestamp(String payload, Long timestamp) { |
241 | 297 | return payload; |
242 | 298 | } |
243 | 299 |
|
244 | | - /** |
245 | | - * In many cases, we explicitly set the record version using the topic's offset. |
246 | | - * This version will, in turn, be checked by Elasticsearch and will throw a versioning |
247 | | - * error if the request represents an equivalent or older version of the record. |
248 | | - * |
249 | | - * @param request the request currently being constructed for `record` |
250 | | - * @param record the record to be processed |
251 | | - * @return the (possibly modified) request which was passed in |
252 | | - */ |
253 | | - private DocWriteRequest<?> maybeAddExternalVersioning( |
254 | | - DocWriteRequest<?> request, |
255 | | - SinkRecord record |
256 | | - ) { |
257 | | - if (!config.isDataStream() && !config.shouldIgnoreKey(record.topic())) { |
258 | | - request.versionType(VersionType.EXTERNAL); |
259 | | - if (config.hasExternalVersionHeader()) { |
260 | | - final Header versionHeader = record.headers().lastWithName(config.externalVersionHeader()); |
261 | | - final byte[] versionValue = HEADER_CONVERTER.fromConnectHeader( |
262 | | - record.topic(), |
263 | | - versionHeader.key(), |
264 | | - versionHeader.schema(), |
265 | | - versionHeader.value() |
266 | | - ); |
267 | | - try { |
268 | | - //fromConnectHeader byte output is UTF_8 |
269 | | - request.version(Long.parseLong(new String(versionValue, StandardCharsets.UTF_8))); |
270 | | - } catch (NumberFormatException e) { |
271 | | - throw new ConnectException("Error converting to long: " |
272 | | - + new String(versionValue, StandardCharsets.UTF_8), e); |
273 | | - } |
274 | | - } else { |
275 | | - request.version(record.kafkaOffset()); |
276 | | - } |
277 | | - } |
278 | | - |
279 | | - return request; |
280 | | - } |
281 | | - |
282 | 300 | // We need to pre process the Kafka Connect schema before converting to JSON as Elasticsearch |
283 | 301 | // expects a different JSON format from the current JSON converter provides. Rather than |
284 | 302 | // completely rewrite a converter for Elasticsearch, we will refactor the JSON converter to |
|
0 commit comments