Skip to content
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

fix: fix NPE when printing records with empty value (MINOR) #3470

Merged
merged 3 commits into from
Oct 3, 2019
Merged

fix: fix NPE when printing records with empty value (MINOR) #3470

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
284dab9
fix: fix NPE when printing records with empty value
vcrfxia Oct 3, 2019
007ca4b
fix: feedback
vcrfxia Oct 3, 2019
300359a
fix: also prevent NPE on null bytes
vcrfxia Oct 3, 2019
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
5 changes: 4 additions & 1 deletion ...rest-app/src/main/java/io/confluent/ksql/rest/server/resources/streaming/TopicStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ public List<String> format(final ConsumerRecords<String, Bytes> records) {
.stream(records.records(topicName).spliterator(), false)
.filter(Objects::nonNull)
.filter(r -> r.value() != null)
.filter(r -> r.value().get().length != 0)
.map((record) -> {
if (formatter == null) {
formatter = getFormatter(record);
Expand Down Expand Up @@ -179,7 +180,9 @@ public String print(final ConsumerRecord<String, Bytes> record)

objectNode.put(SchemaUtil.ROWTIME_NAME.name(), record.timestamp());
objectNode.put(SchemaUtil.ROWKEY_NAME.name(), key);
objectNode.setAll((ObjectNode) jsonNode);
if (jsonNode != null) {
Copy link
Contributor Author

Choose a reason for hiding this comment

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

This fix isn't necessary since the above addition (on line 71) takes care of it, but I figure the redundancy can't hurt since we don't want to hit an NPE if something else we don't know about slips through?

Copy link
Contributor

Choose a reason for hiding this comment

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

I prefer omitting this fix because jsonNode being null could be a symptom for a different issue (assuming we expect it to be non-null)

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Done.

objectNode.setAll((ObjectNode) jsonNode);
}

final StringWriter stringWriter = new StringWriter();
objectMapper.writeValue(stringWriter, objectNode);
Expand Down
70 changes: 42 additions & 28 deletions ...-app/src/test/java/io/confluent/ksql/rest/server/resources/streaming/TopicStreamTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,15 @@

public class TopicStreamTest {

private static final String TOPIC_NAME = "some-topic";

private SchemaRegistryClient schemaRegistryClient;
private RecordFormatter formatter;

@Before
public void setUp() {
schemaRegistryClient = mock(SchemaRegistryClient.class);
formatter = new RecordFormatter(schemaRegistryClient, TOPIC_NAME);
}

@Test
Expand All @@ -83,7 +87,7 @@ public void shouldMatchAvroFormatter() throws Exception {
final byte[] avroData = serializeAvroRecord(avroRecord);

// When:
final Result result = getFormatter(avroData);
final Result result = getFormattedResult(avroData);

// Then:
assertThat(result.format, is(Format.AVRO));
Expand All @@ -97,7 +101,7 @@ public void shouldNotMatchAvroFormatter() {
final String notAvro = "test-data";

// When:
final Result result = getFormatter(notAvro);
final Result result = getFormattedResult(notAvro);

// Then:
assertThat(result.format, is(not(Format.AVRO)));
Expand All @@ -114,7 +118,7 @@ public void shouldFormatJson() {
"}";

// When:
final Result result = getFormatter(json);
final Result result = getFormattedResult(json);

// Then:
assertThat(result.format, is(Format.JSON));
Expand All @@ -133,7 +137,7 @@ public void shouldNotMatchJsonFormatter() {
"}";

// When:
final Result result = getFormatter(notJson);
final Result result = getFormattedResult(notJson);

// Then:
assertThat(result.format, is(not(Format.JSON)));
Expand All @@ -147,25 +151,34 @@ public void shouldMatchStringFormatWithOneColumnValues() {
final String stringValue = "v1";

// When:
final Result result = getFormatter(stringValue);
final Result result = getFormattedResult(stringValue);

// Then:
assertThat(result.format, is(Format.STRING));
}

@Test
public void shouldFilterNullValues() {
// Given:
replay(schemaRegistryClient);

final ConsumerRecord<String, Bytes> record = new ConsumerRecord<>(
"some-topic", 1, 1, "key", null);
final RecordFormatter formatter =
new RecordFormatter(schemaRegistryClient, "some-topic");
final ConsumerRecords<String, Bytes> records = new ConsumerRecords<>(
ImmutableMap.of(new TopicPartition("some-topic", 1),
ImmutableList.of(record)));
// When:
final List<String> formatted = getFormattedRecord(null);

// Then:
assertThat(formatted, empty());
}

@Test
public void shouldFilterEmptyValues() {
// Given:
replay(schemaRegistryClient);

assertThat(formatter.format(records), empty());
// When:
final List<String> formatted = getFormattedRecord(Bytes.EMPTY);

// Then:
assertThat(formatted, empty());
}

@Test
Expand All @@ -174,34 +187,35 @@ public void shouldHandleNullValuesFromSTRINGPrint() throws IOException {
SimpleDateFormat.getDateTimeInstance(3, 1, Locale.getDefault());

final ConsumerRecord<String, Bytes> record = new ConsumerRecord<>(
"some-topic", 1, 1, "key", null);
TOPIC_NAME, 1, 1, "key", null);

final String formatted =
Format.STRING.maybeGetFormatter(
"some-topic", record, null, dateFormat).get().print(record);
TOPIC_NAME, record, null, dateFormat).get().print(record);

assertThat(formatted, endsWith(", key , NULL\n"));
}

private Result getFormatter(final String data) {
return getFormatter(data.getBytes(StandardCharsets.UTF_8));
private Result getFormattedResult(final String data) {
return getFormattedResult(data.getBytes(StandardCharsets.UTF_8));
}

private Result getFormatter(final byte[] data) {
final ConsumerRecord<String, Bytes> record = new ConsumerRecord<>(
"some-topic", 1, 1, "key", new Bytes(data));
private Result getFormattedResult(final byte[] data) {
final List<String> formatted = getFormattedRecord(data);
assertThat("Only expect one line", formatted, hasSize(1));

final RecordFormatter formatter =
new RecordFormatter(schemaRegistryClient, "some-topic");
return new Result(formatter.getFormat(), formatted.get(0));
}

final ConsumerRecords<String, Bytes> records = new ConsumerRecords<>(
ImmutableMap.of(new TopicPartition("some-topic", 1),
ImmutableList.of(record)));
private List<String> getFormattedRecord(final byte[] data) {
final ConsumerRecord<String, Bytes> record = new ConsumerRecord<>(
TOPIC_NAME, 1, 1, "key", new Bytes(data));

final List<String> formatted = formatter.format(records);
assertThat("Only expect one line", formatted, hasSize(1));
final ConsumerRecords<String, Bytes> records = new ConsumerRecords<>(
ImmutableMap.of(new TopicPartition(TOPIC_NAME, 1),
ImmutableList.of(record)));

return new Result(formatter.getFormat(), formatted.get(0));
return formatter.format(records);
}

@SuppressWarnings("SameParameterValue")
Expand Down