{
if (!(data instanceof ZNRecord))
{
// null is NOT an instance of any class
LOG.error("Input object must be of type ZNRecord but it is " + data + ". Will not write to zk");
throw new HelixException("Input object is not of type ZNRecord (was " + data + ")");
}
// apply retention policy on list field
ZNRecord record = (ZNRecord) data;
int max = getListFieldBound(record);
if (max < Integer.MAX_VALUE)
{
Map<String, List<String>> listMap = record.getListFields();
for (String key : listMap.keySet())
{
List<String> list = listMap.get(key);
if (list.size() > max)
{
listMap.put(key, list.subList(0, max));
}
}
}
StringWriter sw = new StringWriter();
try
{
JsonFactory f = new JsonFactory();
JsonGenerator g = f.createJsonGenerator(sw);
g.writeStartObject();
// write id field
g.writeRaw("\n ");
g.writeStringField("id", record.getId());
// write simepleFields
g.writeRaw("\n ");
g.writeObjectFieldStart("simpleFields");
for (String key : record.getSimpleFields().keySet())
{
g.writeRaw("\n ");
g.writeStringField(key, record.getSimpleField(key));
}
g.writeRaw("\n ");
g.writeEndObject(); // for simpleFields
// write listFields
g.writeRaw("\n ");
g.writeObjectFieldStart("listFields");
for (String key : record.getListFields().keySet())
{
// g.writeStringField(key, record.getListField(key).toString());
// g.writeObjectFieldStart(key);
g.writeRaw("\n ");
g.writeArrayFieldStart(key);
List<String> list = record.getListField(key);
for (String listValue : list)
{
g.writeString(listValue);
}
// g.writeEndObject();
g.writeEndArray();
}
g.writeRaw("\n ");
g.writeEndObject(); // for listFields
// write mapFields
g.writeRaw("\n ");
g.writeObjectFieldStart("mapFields");
for (String key : record.getMapFields().keySet())
{
// g.writeStringField(key, record.getMapField(key).toString());
g.writeRaw("\n ");
g.writeObjectFieldStart(key);
Map<String, String> map = record.getMapField(key);
for (String mapKey : map.keySet())
{
g.writeRaw("\n ");
g.writeStringField(mapKey, map.get(mapKey));
}
g.writeRaw("\n ");
g.writeEndObject();
}
g.writeRaw("\n ");
g.writeEndObject(); // for mapFields
g.writeRaw("\n");
g.writeEndObject(); // for whole znrecord
// important: will force flushing of output, close underlying output
// stream
g.close();
}
catch (Exception e)
{
LOG.error("Exception during data serialization. Will not write to zk. Data (first 1k): "
+ sw.toString().substring(0, 1024), e);
throw new HelixException(e);
}
// check size
if (sw.toString().getBytes().length > ZNRecord.SIZE_LIMIT)
{
LOG.error("Data size larger than 1M, ZNRecord.id: " + record.getId()
+ ". Will not write to zk. Data (first 1k): " + sw.toString().substring(0, 1024));
throw new HelixException("Data size larger than 1M, ZNRecord.id: " + record.getId());
}
return sw.toString().getBytes();
}