You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "shibd (via GitHub)" <gi...@apache.org> on 2023/07/06 12:17:55 UTC
[GitHub] [pulsar] shibd opened a new pull request, #20741: Fix: elastic-search sink connector not support JSON.String schema.
shibd opened a new pull request, #20741:
URL: https://github.com/apache/pulsar/pull/20741
### Motivation
If a producer use `Schema.JSON(String.class)` schema, es sink connector will throw `ClassCastException`.
1. Create a producer.
```java
Producer<String> producer = client.newProducer(Schema.JSON(String.class))
.topic("test-es")
.create();
MessageId msgID = producer.send({\"a\":1});
```
2. When es sink connector handle this message, it will throw `ClassCastException`.
```
java.lang.ClassCastException: class java.lang.String cannot be cast to class com.fasterxml.jackson.databind.JsonNode (java.lang.String is in module java.base of loader 'bootstrap'; com.fasterxml.jackson.databind.JsonNode is in unnamed module of loader 'app')
```
Exception throws on here:
https://github.com/apache/pulsar/blob/593fcb87b6d3bd6401eb4ffa4c25d8b3f724e49d/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java#L382
The root cause is when the schema is not `RECORD` type, it will not transfer to `JsonNode`.
https://github.com/apache/pulsar/blob/af1360fb167c1f9484fda5771df3ea9b21d1440b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java#L204-L220
### Modifications
- When `navtiveObject` is String, try to transfer it to JsonNode.
### Verifying this change
- Add `sendJsonStringSchemaTest ` to cover it.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shibd merged pull request #20741: [fix] [io] elastic-search sink connector not support JSON.String schema.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd merged PR #20741:
URL: https://github.com/apache/pulsar/pull/20741
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on pull request #20741: Fix: elastic-search sink connector not support JSON.String schema.
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #20741:
URL: https://github.com/apache/pulsar/pull/20741#issuecomment-1623958653
@shibd
Could you add the labels which branches should cherry-pick into?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] congbobo184 commented on a diff in pull request #20741: [fix] [io] elastic-search sink connector not support JSON.String schema.
Posted by "congbobo184 (via GitHub)" <gi...@apache.org>.
congbobo184 commented on code in PR #20741:
URL: https://github.com/apache/pulsar/pull/20741#discussion_r1258086004
##########
pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java:
##########
@@ -243,6 +243,46 @@ public Schema answer(InvocationOnMock invocation) throws Throwable {
verify(mockRecord, times(1)).ack();
}
+ @Test
+ public final void sendJsonStringSchemaTest() throws Exception {
+
+ when(mockRecord.getMessage()).thenAnswer(new Answer<Optional<Message<String>>>() {
+ @Override
+ public Optional<Message<String>> answer(InvocationOnMock invocation) throws Throwable {
+ final MessageImpl mock = mock(MessageImpl.class);
+ when(mock.getData()).thenReturn("{\"a\":1}".getBytes(StandardCharsets.UTF_8));
+ return Optional.of(mock);
+ }
+ });
+
+ when(mockRecord.getKey()).thenAnswer(new Answer<Optional<String>>() {
+ public Optional<String> answer(InvocationOnMock invocation) throws Throwable {
+ return Optional.empty();
+ }
+ });
+
+ GenericRecord genericRecord = mock(GenericRecord.class);
+ when(genericRecord.getNativeObject()).thenReturn("{\"a\":1}");
+ when(genericRecord.getSchemaType()).thenReturn(SchemaType.JSON);
+ when(mockRecord.getValue()).thenAnswer(new Answer<GenericRecord>() {
+ public GenericRecord answer(InvocationOnMock invocation) throws Throwable {
+ return genericRecord;
+ }
+ });
+
+ when(mockRecord.getSchema()).thenAnswer(new Answer<Schema>() {
+ public Schema answer(InvocationOnMock invocation) throws Throwable {
+ return Schema.JSON(String.class);
+ }
+ });
+
+ map.put("indexName", "test-index");
+ map.put("schemaEnable", "true");
+ sink.open(map, mockSinkContext);
+ sink.write(mockRecord);
+ verify(mockRecord, times(1)).ack();
Review Comment:
may be we should add a test for the exception that you added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] poorbarcode commented on pull request #20741: [fix] [io] elastic-search sink connector not support JSON.String schema.
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on PR #20741:
URL: https://github.com/apache/pulsar/pull/20741#issuecomment-1623963489
/pulsarbot rerun-failure-checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] shibd commented on a diff in pull request #20741: [fix] [io] elastic-search sink connector not support JSON.String schema.
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #20741:
URL: https://github.com/apache/pulsar/pull/20741#discussion_r1259389301
##########
pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java:
##########
@@ -243,6 +243,46 @@ public Schema answer(InvocationOnMock invocation) throws Throwable {
verify(mockRecord, times(1)).ack();
}
+ @Test
+ public final void sendJsonStringSchemaTest() throws Exception {
+
+ when(mockRecord.getMessage()).thenAnswer(new Answer<Optional<Message<String>>>() {
+ @Override
+ public Optional<Message<String>> answer(InvocationOnMock invocation) throws Throwable {
+ final MessageImpl mock = mock(MessageImpl.class);
+ when(mock.getData()).thenReturn("{\"a\":1}".getBytes(StandardCharsets.UTF_8));
+ return Optional.of(mock);
+ }
+ });
+
+ when(mockRecord.getKey()).thenAnswer(new Answer<Optional<String>>() {
+ public Optional<String> answer(InvocationOnMock invocation) throws Throwable {
+ return Optional.empty();
+ }
+ });
+
+ GenericRecord genericRecord = mock(GenericRecord.class);
+ when(genericRecord.getNativeObject()).thenReturn("{\"a\":1}");
+ when(genericRecord.getSchemaType()).thenReturn(SchemaType.JSON);
+ when(mockRecord.getValue()).thenAnswer(new Answer<GenericRecord>() {
+ public GenericRecord answer(InvocationOnMock invocation) throws Throwable {
+ return genericRecord;
+ }
+ });
+
+ when(mockRecord.getSchema()).thenAnswer(new Answer<Schema>() {
+ public Schema answer(InvocationOnMock invocation) throws Throwable {
+ return Schema.JSON(String.class);
+ }
+ });
+
+ map.put("indexName", "test-index");
+ map.put("schemaEnable", "true");
+ sink.open(map, mockSinkContext);
+ sink.write(mockRecord);
+ verify(mockRecord, times(1)).ack();
Review Comment:
Thanks, added a test for converting JSON exception, PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org