Skip to content

Commit 49b250d

Browse files
dgray16dgray16
authored
[bq] Break hard dependency on Jackson for JSON processing
Signed-off-by: Volodymyr Perebykivskyi <[email protected]>
1 parent 4114825 commit 49b250d

File tree

6 files changed

+52
-83
lines changed

6 files changed

+52
-83
lines changed

‎spring-batch-bigquery/pom.xml‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -100,15 +100,14 @@
100100
<dependency>
101101
<groupId>org.wiremock</groupId>
102102
<artifactId>wiremock-standalone</artifactId>
103-
<version>3.13.0</version>
103+
<version>3.13.1</version>
104104
<scope>test</scope>
105105
</dependency>
106106
<dependency>
107107
<groupId>org.slf4j</groupId>
108108
<artifactId>jul-to-slf4j</artifactId>
109109
<scope>test</scope>
110110
</dependency>
111-
112111
</dependencies>
113112

114113
<build>

‎spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryJsonItemWriter.java‎

Lines changed: 9 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,9 @@
1616

1717
package org.springframework.batch.extensions.bigquery.writer;
1818

19-
import com.fasterxml.jackson.core.JsonProcessingException;
20-
import com.fasterxml.jackson.databind.ObjectMapper;
21-
import com.fasterxml.jackson.databind.ObjectWriter;
2219
import com.google.cloud.bigquery.FormatOptions;
2320
import com.google.cloud.bigquery.Table;
24-
import org.springframework.core.convert.converter.Converter;
21+
import org.springframework.batch.item.json.JsonObjectMarshaller;
2522
import org.springframework.util.Assert;
2623
import org.springframework.util.ObjectUtils;
2724

@@ -42,41 +39,27 @@ public class BigQueryJsonItemWriter<T> extends BigQueryBaseItemWriter<T> {
4239

4340
private static final String LF = "\n";
4441

45-
private Converter<T, String> rowMapper;
46-
private ObjectWriter objectWriter;
47-
private Class<?> itemClass;
42+
private JsonObjectMarshaller<T> marshaller;
4843

4944
@Override
5045
protected void doInitializeProperties(List<? extends T> items) {
51-
if (this.itemClass == null) {
52-
T firstItem = items.stream().findFirst().orElseThrow(() -> {
53-
logger.warn("Class type was not found");
54-
return new IllegalStateException("Class type was not found");
55-
});
56-
this.itemClass = firstItem.getClass();
57-
58-
if (this.rowMapper == null) {
59-
this.objectWriter = new ObjectMapper().writerFor(this.itemClass);
60-
}
61-
62-
logger.debug("Writer setup is completed");
63-
}
46+
// Unused
6447
}
6548

6649
/**
6750
* Converter that transforms a single row into a {@link String}.
6851
*
69-
* @param rowMapper your JSON row mapper
52+
* @param marshaller your JSON mapper
7053
*/
71-
public void setRowMapper(Converter<T, String> rowMapper) {
72-
this.rowMapper = rowMapper;
54+
public void setMarshaller(JsonObjectMarshaller<T> marshaller) {
55+
this.marshaller = marshaller;
7356
}
7457

7558
@Override
7659
protected List<byte[]> convertObjectsToByteArrays(List<? extends T> items) {
7760
return items
7861
.stream()
79-
.map(this::mapItemToJson)
62+
.map(marshaller::marshal)
8063
.filter(Predicate.not(ObjectUtils::isEmpty))
8164
.map(this::convertToNdJson)
8265
.map(row -> row.getBytes(StandardCharsets.UTF_8))
@@ -85,6 +68,8 @@ protected List<byte[]> convertObjectsToByteArrays(List<? extends T> items) {
8568

8669
@Override
8770
protected void performFormatSpecificChecks() {
71+
Assert.notNull(this.marshaller, "Marshaller is mandatory");
72+
8873
Table table = getTable();
8974

9075
if (Boolean.TRUE.equals(writeChannelConfig.getAutodetect())) {
@@ -106,16 +91,6 @@ protected void performFormatSpecificChecks() {
10691
Assert.isTrue(Objects.equals(format, super.writeChannelConfig.getFormat()), "Only %s format is allowed".formatted(format));
10792
}
10893

109-
private String mapItemToJson(T t) {
110-
try {
111-
return rowMapper == null ? objectWriter.writeValueAsString(t) : rowMapper.convert(t);
112-
}
113-
catch (JsonProcessingException e) {
114-
logger.error("Error during processing of the line: ", e);
115-
return null;
116-
}
117-
}
118-
11994
/**
12095
* BigQuery uses <a href="https://github.com/ndjson/ndjson-spec">ndjson</a>.
12196
* It is expected that to pass here JSON line generated by

‎spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryJsonItemWriterBuilder.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import com.google.cloud.bigquery.Job;
2222
import com.google.cloud.bigquery.WriteChannelConfiguration;
2323
import org.springframework.batch.extensions.bigquery.writer.BigQueryJsonItemWriter;
24-
import org.springframework.core.convert.converter.Converter;
24+
import org.springframework.batch.item.json.JsonObjectMarshaller;
2525

2626
import java.util.function.Consumer;
2727

@@ -35,8 +35,7 @@
3535
*/
3636
public class BigQueryJsonItemWriterBuilder<T> {
3737

38-
private Converter<T, String> rowMapper;
39-
38+
private JsonObjectMarshaller<T> marshaller;
4039
private Consumer<Job> jobConsumer;
4140
private DatasetInfo datasetInfo;
4241
private WriteChannelConfiguration writeChannelConfig;
@@ -45,12 +44,12 @@ public class BigQueryJsonItemWriterBuilder<T> {
4544
/**
4645
* Converts your DTO into a {@link String}.
4746
*
48-
* @param rowMapper your mapping
47+
* @param marshaller your mapper
4948
* @return {@link BigQueryJsonItemWriter}
50-
* @see BigQueryJsonItemWriter#setRowMapper(Converter)
49+
* @see BigQueryJsonItemWriter#setMarshaller(JsonObjectMarshaller)
5150
*/
52-
public BigQueryJsonItemWriterBuilder<T> rowMapper(Converter<T, String> rowMapper) {
53-
this.rowMapper = rowMapper;
51+
public BigQueryJsonItemWriterBuilder<T> marshaller(JsonObjectMarshaller<T> marshaller) {
52+
this.marshaller = marshaller;
5453
return this;
5554
}
5655

@@ -110,7 +109,7 @@ public BigQueryJsonItemWriterBuilder<T> bigQuery(BigQuery bigQuery) {
110109
public BigQueryJsonItemWriter<T> build() {
111110
BigQueryJsonItemWriter<T> writer = new BigQueryJsonItemWriter<>();
112111

113-
writer.setRowMapper(this.rowMapper);
112+
writer.setMarshaller(this.marshaller);
114113
writer.setWriteChannelConfig(this.writeChannelConfig);
115114
writer.setJobConsumer(this.jobConsumer);
116115
writer.setBigQuery(this.bigQuery);

‎spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/emulator/writer/BigQueryEmulatorJsonItemWriterTest.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.springframework.batch.extensions.bigquery.writer.BigQueryJsonItemWriter;
3232
import org.springframework.batch.extensions.bigquery.writer.builder.BigQueryJsonItemWriterBuilder;
3333
import org.springframework.batch.item.Chunk;
34+
import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
3435

3536
import java.util.stream.Stream;
3637

@@ -52,6 +53,7 @@ void testWrite(String table, boolean autodetect) throws Exception {
5253
BigQueryJsonItemWriter<PersonDto> writer = new BigQueryJsonItemWriterBuilder<PersonDto>()
5354
.bigQuery(bigQuery)
5455
.writeChannelConfig(channelConfig)
56+
.marshaller(new JacksonJsonObjectMarshaller<>())
5557
.build();
5658
writer.afterPropertiesSet();
5759

‎spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/BigQueryJsonItemWriterTest.java‎

Lines changed: 26 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616

1717
package org.springframework.batch.extensions.bigquery.unit.writer;
1818

19-
import com.fasterxml.jackson.core.JsonFactory;
20-
import com.fasterxml.jackson.databind.ObjectWriter;
2119
import com.google.cloud.bigquery.BigQuery;
2220
import com.google.cloud.bigquery.Field;
2321
import com.google.cloud.bigquery.FormatOptions;
@@ -36,7 +34,9 @@
3634
import org.springframework.batch.extensions.bigquery.common.TestConstants;
3735
import org.springframework.batch.extensions.bigquery.unit.base.AbstractBigQueryTest;
3836
import org.springframework.batch.extensions.bigquery.writer.BigQueryJsonItemWriter;
39-
import org.springframework.core.convert.converter.Converter;
37+
import org.springframework.batch.item.json.GsonJsonObjectMarshaller;
38+
import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
39+
import org.springframework.batch.item.json.JsonObjectMarshaller;
4040

4141
import java.lang.invoke.MethodHandles;
4242
import java.util.List;
@@ -47,34 +47,15 @@ class BigQueryJsonItemWriterTest extends AbstractBigQueryTest {
4747
private static final TableId TABLE_ID = TableId.of("1", "2");
4848

4949
@Test
50-
void testDoInitializeProperties() throws IllegalAccessException, NoSuchFieldException {
51-
TestWriter writer = new TestWriter();
52-
List<PersonDto> items = TestConstants.CHUNK.getItems();
53-
MethodHandles.Lookup handle = MethodHandles.privateLookupIn(BigQueryJsonItemWriter.class, MethodHandles.lookup());
54-
55-
// Exception
56-
Assertions.assertThrows(IllegalStateException.class, () -> writer.testInitializeProperties(List.of()));
57-
58-
// No exception
59-
writer.testInitializeProperties(items);
60-
Assertions.assertEquals(
61-
PersonDto.class.getSimpleName(),
62-
((Class<PersonDto>) handle.findVarHandle(BigQueryJsonItemWriter.class, "itemClass", Class.class).get(writer)).getSimpleName()
63-
);
64-
ObjectWriter objectWriter = (ObjectWriter) handle.findVarHandle(BigQueryJsonItemWriter.class, "objectWriter", ObjectWriter.class).get(writer);
65-
Assertions.assertInstanceOf(JsonFactory.class, objectWriter.getFactory());
66-
}
67-
68-
@Test
69-
void testSetRowMapper() throws IllegalAccessException, NoSuchFieldException {
50+
void testSetMarshaller() throws IllegalAccessException, NoSuchFieldException {
7051
BigQueryJsonItemWriter<PersonDto> reader = new BigQueryJsonItemWriter<>();
7152
MethodHandles.Lookup handle = MethodHandles.privateLookupIn(BigQueryJsonItemWriter.class, MethodHandles.lookup());
72-
Converter<PersonDto, String> expected = source -> null;
53+
JsonObjectMarshaller<PersonDto> expected = new JacksonJsonObjectMarshaller<>();
7354

74-
reader.setRowMapper(expected);
55+
reader.setMarshaller(expected);
7556

76-
Converter<PersonDto, String> actual = (Converter<PersonDto, String>) handle
77-
.findVarHandle(BigQueryJsonItemWriter.class, "rowMapper", Converter.class)
57+
JsonObjectMarshaller<PersonDto> actual = (JsonObjectMarshaller<PersonDto>) handle
58+
.findVarHandle(BigQueryJsonItemWriter.class, "marshaller", JsonObjectMarshaller.class)
7859
.get(reader);
7960

8061
Assertions.assertEquals(expected, actual);
@@ -83,14 +64,23 @@ void testSetRowMapper() throws IllegalAccessException, NoSuchFieldException {
8364
@Test
8465
void testConvertObjectsToByteArrays() {
8566
TestWriter writer = new TestWriter();
67+
writer.setMarshaller(new JacksonJsonObjectMarshaller<>());
8668

8769
// Empty
8870
Assertions.assertTrue(writer.testConvert(List.of()).isEmpty());
8971

9072
// Not empty
91-
writer.setRowMapper(Record::toString);
73+
writer.setMarshaller(Record::toString);
9274
List<byte[]> actual = writer.testConvert(TestConstants.CHUNK.getItems());
93-
List<byte[]> expected = TestConstants.CHUNK.getItems().stream().map(PersonDto::toString).map(s -> s.concat("\n")).map(String::getBytes).toList();
75+
76+
List<byte[]> expected = TestConstants.CHUNK
77+
.getItems()
78+
.stream()
79+
.map(PersonDto::toString)
80+
.map(s -> s.concat("\n"))
81+
.map(String::getBytes)
82+
.toList();
83+
9484
Assertions.assertEquals(expected.size(), actual.size());
9585

9686
for (int i = 0; i < actual.size(); i++) {
@@ -112,10 +102,15 @@ void testPerformFormatSpecificChecks() {
112102
BigQuery bigQuery = prepareMockedBigQuery();
113103
Mockito.when(bigQuery.getTable(Mockito.any(TableId.class))).thenReturn(table);
114104

105+
// marshaller
106+
IllegalArgumentException actual = Assertions.assertThrows(IllegalArgumentException.class, writer::testPerformFormatSpecificChecks);
107+
Assertions.assertEquals("Marshaller is mandatory", actual.getMessage());
108+
115109
// schema
110+
writer.setMarshaller(new JacksonJsonObjectMarshaller<>());
116111
writer.setBigQuery(bigQuery);
117112
writer.setWriteChannelConfig(WriteChannelConfiguration.of(TABLE_ID, FormatOptions.csv()));
118-
IllegalArgumentException actual = Assertions.assertThrows(IllegalArgumentException.class, writer::testPerformFormatSpecificChecks);
113+
actual = Assertions.assertThrows(IllegalArgumentException.class, writer::testPerformFormatSpecificChecks);
119114
Assertions.assertEquals("Schema must be provided", actual.getMessage());
120115

121116
// schema equality
@@ -144,6 +139,7 @@ void testPerformFormatSpecificChecks_Format(FormatOptions formatOptions) {
144139

145140
TestWriter writer = new TestWriter();
146141
writer.setBigQuery(bigQuery);
142+
writer.setMarshaller(new GsonJsonObjectMarshaller<>());
147143

148144
writer.setWriteChannelConfig(WriteChannelConfiguration.newBuilder(TABLE_ID).setAutodetect(true).setFormatOptions(formatOptions).build());
149145
IllegalArgumentException actual = Assertions.assertThrows(IllegalArgumentException.class, writer::testPerformFormatSpecificChecks);
@@ -164,9 +160,6 @@ static Stream<FormatOptions> invalidFormats() {
164160
}
165161

166162
private static final class TestWriter extends BigQueryJsonItemWriter<PersonDto> {
167-
public void testInitializeProperties(List<PersonDto> items) {
168-
doInitializeProperties(items);
169-
}
170163

171164
public List<byte[]> testConvert(List<PersonDto> items) {
172165
return convertObjectsToByteArrays(items);

‎spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryJsonItemWriterBuilderTests.java‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,8 @@
3030
import org.springframework.batch.extensions.bigquery.writer.BigQueryBaseItemWriter;
3131
import org.springframework.batch.extensions.bigquery.writer.BigQueryJsonItemWriter;
3232
import org.springframework.batch.extensions.bigquery.writer.builder.BigQueryJsonItemWriterBuilder;
33-
import org.springframework.core.convert.converter.Converter;
33+
import org.springframework.batch.item.json.JacksonJsonObjectMarshaller;
34+
import org.springframework.batch.item.json.JsonObjectMarshaller;
3435

3536
import java.lang.invoke.MethodHandles;
3637
import java.util.function.Consumer;
@@ -42,7 +43,7 @@ void testBuild() throws IllegalAccessException, NoSuchFieldException {
4243
MethodHandles.Lookup jsonWriterHandle = MethodHandles.privateLookupIn(BigQueryJsonItemWriter.class, MethodHandles.lookup());
4344
MethodHandles.Lookup baseWriterHandle = MethodHandles.privateLookupIn(BigQueryBaseItemWriter.class, MethodHandles.lookup());
4445

45-
Converter<PersonDto, String> rowMapper = source -> "";
46+
JsonObjectMarshaller<PersonDto> marshaller = new JacksonJsonObjectMarshaller<>();
4647
DatasetInfo datasetInfo = DatasetInfo.newBuilder(TestConstants.DATASET).setLocation("europe-west-2").build();
4748
Consumer<Job> jobConsumer = job -> {};
4849
BigQuery mockedBigQuery = prepareMockedBigQuery();
@@ -53,7 +54,7 @@ void testBuild() throws IllegalAccessException, NoSuchFieldException {
5354
.build();
5455

5556
BigQueryJsonItemWriter<PersonDto> writer = new BigQueryJsonItemWriterBuilder<PersonDto>()
56-
.rowMapper(rowMapper)
57+
.marshaller(marshaller)
5758
.writeChannelConfig(writeConfiguration)
5859
.jobConsumer(jobConsumer)
5960
.bigQuery(mockedBigQuery)
@@ -62,8 +63,8 @@ void testBuild() throws IllegalAccessException, NoSuchFieldException {
6263

6364
Assertions.assertNotNull(writer);
6465

65-
Converter<PersonDto, String> actualRowMapper = (Converter<PersonDto, String>) jsonWriterHandle
66-
.findVarHandle(BigQueryJsonItemWriter.class, "rowMapper", Converter.class)
66+
JsonObjectMarshaller<PersonDto> actualMarshaller = (JsonObjectMarshaller<PersonDto>) jsonWriterHandle
67+
.findVarHandle(BigQueryJsonItemWriter.class, "marshaller", JsonObjectMarshaller.class)
6768
.get(writer);
6869

6970
WriteChannelConfiguration actualWriteChannelConfig = (WriteChannelConfiguration) jsonWriterHandle
@@ -82,7 +83,7 @@ void testBuild() throws IllegalAccessException, NoSuchFieldException {
8283
.findVarHandle(BigQueryJsonItemWriter.class, "datasetInfo", DatasetInfo.class)
8384
.get(writer);
8485

85-
Assertions.assertEquals(rowMapper, actualRowMapper);
86+
Assertions.assertEquals(marshaller, actualMarshaller);
8687
Assertions.assertEquals(writeConfiguration, actualWriteChannelConfig);
8788
Assertions.assertEquals(jobConsumer, actualJobConsumer);
8889
Assertions.assertEquals(mockedBigQuery, actualBigQuery);

0 commit comments

Comments
 (0)