Skip to content

Commit

Permalink
Merge pull request #13 from sowmya-dixit/secor-0.29
Browse files Browse the repository at this point in the history 
TG-928: Channel backup fix to handle empty partition value
  • Loading branch information
@SanthoshVasabhaktula
SanthoshVasabhaktula authored May 7, 2021
2 parents 8b6172f + 24999a0 commit dac0dda
Show file tree
Hide file tree
Showing 2 changed files with 26 additions and 6 deletions.
14 changes: 8 additions & 6 deletions src/main/java/com/pinterest/secor/parser/ChannelDateMessageParser.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -137,29 +137,31 @@ private String getPrefix(String prefixIdentifier) {
}

private String getChannel(JSONObject jsonObject) {
String rawChannelStr = "";
String channelStr = "";
String finalChannelStr = "";
String[] channelIdentifier = mConfig.getMessageChannelIdentifier();
if (channelIdentifier.length > 0) {
try {
// If channel identifier is common for both raw and summary events
if (channelIdentifier.length == 1) {
rawChannelStr = JsonPath.parse(jsonObject).read("$." + channelIdentifier[0], String.class);
channelStr = JsonPath.parse(jsonObject).read("$." + channelIdentifier[0], String.class);
}
// If channel identifier is different for raw and summary events. Ex: context.channel/dimensions.channel
else {
try {
rawChannelStr = JsonPath.parse(jsonObject).read("$." + channelIdentifier[0], String.class);
channelStr = JsonPath.parse(jsonObject).read("$." + channelIdentifier[0], String.class);
}
catch (PathNotFoundException e) {
LOG.warn("Unable to get path: " + e.getMessage());
rawChannelStr = JsonPath.parse(jsonObject).read("$." + channelIdentifier[1], String.class);
channelStr = JsonPath.parse(jsonObject).read("$." + channelIdentifier[1], String.class);
}
}
} catch (PathNotFoundException e) {
LOG.warn("Unable to get path: " + e.getMessage());
rawChannelStr = "others";
channelStr = "others";
}
}
return rawChannelStr.replaceAll(channelScrubRegex, "");
if (channelStr.isEmpty()) finalChannelStr = "others"; else finalChannelStr = channelStr;
return finalChannelStr.replaceAll(channelScrubRegex, "");
}
}
18 changes: 18 additions & 0 deletions src/test/java/com/pinterest/secor/parser/ChannelDateMessageParserTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ public class ChannelDateMessageParserTest extends TestCase {
private Message mFormat1;
private Message mFormat2;
private Message mFormat3;
private Message mFormat4;
private Message mInvalidDate;
private Message mISOFormat;
private Message mNanosecondISOFormat;
Expand Down Expand Up @@ -46,6 +47,11 @@ public void setUp() throws Exception {
.getBytes("UTF-8");
mFormat3 = new Message("test", 0, 0, null, format3, timestamp, null);

byte format4[] = "{\"timestamp\":\"2014-07-30 22:53:20\",\"eid\":\"ME_WORKFLOW_SUMMARY\",\"dimensions\":{\"channel\":\"test-channel\"},\"id\":0,\"guid\":\"0436b17b-e78a-4e82-accf-743bf1f0b884\",\"isActive\":false,\"balance\":\"$3,561.87\",\"picture\":\"http://placehold.it/32x32\",\"age\":23,\"eyeColor\":\"green\",\"name\":\"Mercedes Brewer\",\"gender\":\"female\",\"company\":\"MALATHION\",\"email\":\"[email protected]\",\"phone\":\"+1 (848) 471-3000\",\"address\":\"786 Gilmore Court, Brule, Maryland, 3200\",\"about\":\"Quis nostrud Lorem deserunt esse ut reprehenderit aliqua nisi et sunt mollit est. Cupidatat incididunt minim anim eiusmod culpa elit est dolor ullamco. Aliqua cillum eiusmod ullamco nostrud Lorem sit amet Lorem aliquip esse esse velit.\\r\\n\",\"registered\":\"2014-01-14T13:07:28 +08:00\",\"latitude\":47.672012,\"longitude\":102.788623,\"tags\":[\"amet\",\"amet\",\"dolore\",\"eu\",\"qui\",\"fugiat\",\"laborum\"],\"friends\":[{\"id\":0,\"name\":\"Rebecca Hardy\"},{\"id\":1,\"name\":\"Sutton Briggs\"},{\"id\":2,\"name\":\"Dena Campos\"}],\"greeting\":\"Hello, Mercedes Brewer! You have 7 unread messages.\",\"favoriteFruit\":\"strawberry\",\"derivedlocationdata\":{\"district\":\"Bengaluru\",\"state\":\"\",\"from\":\"user-profile\"}}"
.getBytes("UTF-8");
mFormat4 = new Message("test", 0, 0, null, format4, timestamp, null);


byte invalidDate[] = "{\"timestamp\":\"11111111\",\"eid\":\"ME_WORKFLOW_SUMMARY\",\"dimensions\":{\"channel\":\"test-channel\"},\"id\":0,\"guid\":\"0436b17b-e78a-4e82-accf-743bf1f0b884\",\"isActive\":false,\"balance\":\"$3,561.87\",\"picture\":\"http://placehold.it/32x32\",\"age\":23,\"eyeColor\":\"green\",\"name\":\"Mercedes Brewer\",\"gender\":\"female\",\"company\":\"MALATHION\",\"email\":\"[email protected]\",\"phone\":\"+1 (848) 471-3000\",\"address\":\"786 Gilmore Court, Brule, Maryland, 3200\",\"about\":\"Quis nostrud Lorem deserunt esse ut reprehenderit aliqua nisi et sunt mollit est. Cupidatat incididunt minim anim eiusmod culpa elit est dolor ullamco. Aliqua cillum eiusmod ullamco nostrud Lorem sit amet Lorem aliquip esse esse velit.\\r\\n\",\"registered\":\"2014-01-14T13:07:28 +08:00\",\"latitude\":47.672012,\"longitude\":102.788623,\"tags\":[\"amet\",\"amet\",\"dolore\",\"eu\",\"qui\",\"fugiat\",\"laborum\"],\"friends\":[{\"id\":0,\"name\":\"Rebecca Hardy\"},{\"id\":1,\"name\":\"Sutton Briggs\"},{\"id\":2,\"name\":\"Dena Campos\"}],\"greeting\":\"Hello, Mercedes Brewer! You have 7 unread messages.\",\"favoriteFruit\":\"strawberry\",\"derivedlocationdata\":{\"district\":\"Bengaluru\",\"state\":\"Karnataka\",\"from\":\"user-profile\"}}"
.getBytes("UTF-8");
mInvalidDate = new Message("test", 0, 0, null, invalidDate, timestamp, null);
Expand Down Expand Up @@ -158,4 +164,16 @@ public void testMessageChannelIdentifier() throws Exception {
assertEquals("raw/test-channel/2014-10-25", new ChannelDateMessageParser(mConfig).extractPartitions(mFormat2)[0]);
}

@Test
public void testEmptyChannelPrefix() throws Exception {
String[] channelIdentifier = {"derivedlocationdata.state"};
Mockito.when(mConfig.getMessageTimestampName()).thenReturn("timestamp");
Mockito.when(mConfig.getPartitionPrefixMapping()).thenReturn("{\"ME_WORKFLOW_SUMMARY\":\"summary\",\"DEFAULT\":\"raw\"}");
Mockito.when(mConfig.getMessageTimestampInputPattern()).thenReturn("yyyy-MM-dd HH:mm:ss");
Mockito.when(mConfig.getString("secor.partition.output_dt_format", "yyyy-MM-dd")).thenReturn("yyyy-MM-dd");
Mockito.when(mConfig.getMessageChannelIdentifier()).thenReturn(channelIdentifier);

assertEquals("summary/others/2014-07-31", new ChannelDateMessageParser(mConfig).extractPartitions(mFormat4)[0]);
}

}

0 comments on commit dac0dda

Please sign in to comment.