Java 类org.apache.avro.generic.GenericData.Record 实例源码

/**
 * Reads in binary Avro-encoded entities using the schema stored in the file
 * and prints them out.
 */
public static void readAvroFile(File file) throws IOException {
    GenericDatumReader datum = new GenericDatumReader();
    DataFileReader reader = new DataFileReader(file, datum);

    GenericData.Record record = new GenericData.Record(reader.getSchema());
    while (reader.hasNext()) {
        reader.next(record);
        System.out.println("Name " + record.get("name") + " on "
                + record.get("Meetup_date") + " attending "
                + record.get("going") + " organized by  "
                + record.get("organizer") + " on  " + record.get("topics"));
    }

    reader.close();
}

@Test
public void publish_shouldThrowAnEncodeTransportExceptionForAnInvalidTransportMessage() throws Exception {

    // expect
    exception.expect(EncodeMessageContentException.class);
    exception.expectCause(IsInstanceOf.<Throwable>instanceOf(ClassCastException.class));
    exception.expectMessage("Failed to publish message: Failed to encode message content");

    // given
    Context context = mock(Context.class);
    when(context.getSchema())
            .thenReturn(new Schema.Parser().parse("{\"name\":\"test\",\"type\":\"record\",\"fields\":[]}"));
    when(context.getDataBuffer()).thenReturn(ByteBuffer.wrap("{}".getBytes()));

    GenericRecordBuilder builder = new GenericRecordBuilder(schema);
    builder.set("testString", Boolean.TRUE);
    Record record = builder.build();

    // when
    publisher.publish(context, record);

}

@Test
public void publish_shouldThrowAnEncodeTransportExceptionForAnInvalidTransportContext() throws Exception {

    // expect
    exception.expect(EncodeTransportException.class);
    exception.expectCause(IsInstanceOf.<Throwable>instanceOf(NullPointerException.class));
    exception.expectMessage("Failed to publish message: Failed to encode transport");

    // given
    Context context = mock(Context.class);
    when(context.getSchema())
            .thenReturn(new Schema.Parser().parse("{\"name\":\"test\",\"type\":\"record\",\"fields\":[]}"));
    when(context.getDataBuffer()).thenReturn(null);

    GenericRecordBuilder builder = new GenericRecordBuilder(schema);
    builder.set("testString", "testString");
    Record record = builder.build();

    // when
    publisher.publish(context, record);

}

@Test
public void publish_shouldPublishTheMessage() throws Exception {

    // given
    Context context = mock(Context.class);
    when(context.getSchema())
            .thenReturn(new Schema.Parser().parse("{\"name\":\"test\",\"type\":\"record\",\"fields\":[]}"));
    when(context.getDataBuffer()).thenReturn(ByteBuffer.wrap("{}".getBytes()));

    GenericRecordBuilder builder = new GenericRecordBuilder(schema);
    builder.set("testString", "test");
    Record record = builder.build();

    // when
    publisher.publish(context, record);

}

/**
 * Reads in binary Avro-encoded entities using a schema that is different
 * from the writer's schema.
 * 
 */
public static void readWithDifferentSchema(File file, Schema newSchema)
        throws IOException {
    GenericDatumReader datum = new GenericDatumReader(newSchema);
    DataFileReader reader = new DataFileReader(file, datum);

    GenericData.Record record = new GenericData.Record(newSchema);
    while (reader.hasNext()) {
        reader.next(record);
        System.out.println("Name " + record.get("name") + " on "
                + record.get("Meetup_date") + " attending "
                + record.get("attendance") + " organized by  "
                + record.get("organizer") 
                + " at  " + record.get("location"));
    }

    reader.close();
}

private static Object[] transformAvroArrayToObjectArray(Array arr) {
  if (arr == null) {
    return new Object[0];
  }
  final Object[] ret = new Object[arr.size()];
  final Iterator iterator = arr.iterator();
  int i = 0;
  while (iterator.hasNext()) {
    Object value = iterator.next();
    if (value instanceof Record) {
      value = ((Record) value).get(0);
    }
    if (value instanceof Utf8) {
      value = ((Utf8) value).toString();
    }
    ret[i++] = value;
  }
  return ret;
}

public static Object[] transformAvroArrayToObjectArray(Array arr, FieldSpec spec) {
  if (arr == null) {
    return new Object[] { getDefaultNullValue(spec) };
  }
  if (arr.size() == 0) {
    return new Object[] { getDefaultNullValue(spec) };
  }

  final Object[] ret = new Object[arr.size()];
  final Iterator iterator = arr.iterator();
  int i = 0;
  while (iterator.hasNext()) {
    Object value = iterator.next();
    if (value instanceof Record) {
      value = ((Record) value).get(0);
    }
    if (value instanceof Utf8) {
      value = ((Utf8) value).toString();
    }
    if (value == null) {
      value = getDefaultNullValue(spec);
    }
    ret[i++] = value;
  }
  return ret;
}

private static Object[] transformAvroArrayToObjectArray(Array arr) {
  if (arr == null) {
    return new Object[0];
  }
  final Object[] ret = new Object[arr.size()];
  final Iterator iterator = arr.iterator();
  int i = 0;
  while (iterator.hasNext()) {
    Object value = iterator.next();
    if (value instanceof Record) {
      value = ((Record) value).get(0);
    }
    if (value instanceof Utf8) {
      value = ((Utf8) value).toString();
    }
    ret[i++] = value;
  }
  return ret;
}

@SuppressWarnings("unchecked")
@Override
public JetstreamEvent decode(byte[] key, byte[] message) {
    ByteArrayInputStream stream = new ByteArrayInputStream(message);
    BinaryDecoder reusedDecoder = decoderHolder.get();
    BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(stream, reusedDecoder);
    if (reusedDecoder == null) {
        decoderHolder.set(decoder);
    }
    Record object;
    try {
        object = reader.read(null, decoder);
        Map<String, Object> m = (Map<String, Object>) object.get(MAP_FIELD_NAME);
        return new JetstreamEvent(m);
    } catch (IOException e) {
        throw new IllegalArgumentException("Can not read the avro message", e);
    }
}

@Test
public void testDefaultValues() throws Exception {
  Schema schema = Defaults.SCHEMA$;

  GenericRecordBuilder builder = new GenericRecordBuilder(schema).set("id", "1234");
  Record record1 = builder.build();

  String json = "{\"id\": \"1234\"}";
  BSONObject object = (BSONObject) JSON.parse(json);
  Record record2 = RecordConverter.toRecord(schema, object, getClass().getClassLoader());

  assertThat(record2, is(record1));
  assertThat(AvroHelper.toSimpleJson(schema, record2), is(AvroHelper.toSimpleJson(schema, record1)));

  assertEquals(record2.get("id"), "1234");
  assertNull(record2.get("s"));
  assertTrue((Boolean) record2.get("b"));
  assertEquals(((Record) record2.get("r")).get("f"), "value");
  assertEquals(((Record) record2.get("r")).get("l"), 1234l);
}

@Test
public void testArrays() throws Exception {
  Schema schema = Arrays.SCHEMA$;

  GenericRecordBuilder builder = new GenericRecordBuilder(schema);
  builder.set("arrays", ImmutableList.of(ImmutableList.of(ImmutableList.of(1, 2, 3),
      ImmutableList.of()), ImmutableList.of(ImmutableList.of(4), ImmutableList.of()),
      ImmutableList.of(ImmutableList.of())));
  Record record1 = builder.build();

  String json = "{\"arrays\": [[[1, 2, 3], []], [[4], []], [[]]]}";
  BSONObject object = (BSONObject) JSON.parse(json);
  Record record2 = RecordConverter.toRecord(schema, object, getClass().getClassLoader());

  assertThat(record2, is(record1));
  assertThat(AvroHelper.toSimpleJson(schema, record2), is(AvroHelper.toSimpleJson(schema, record1)));
}

@Test
public void testEnums() throws Exception {
  Schema schema = Enums.SCHEMA$;

  String avroJson = "{\"enum1\": \"X\", \"enum2\": {\"test.Enum2\": \"A\"}, \"enum3\": {\"null\": null}, \"enum4\": [{\"test.Enum4\": \"SAT\"}, {\"test.Enum4\": \"SUN\"}]}}";
  Decoder decoder = DecoderFactory.get().jsonDecoder(schema, avroJson);
  GenericDatumReader<Record> reader = new GenericDatumReader<Record>(schema);
  Record record1 = reader.read(null, decoder);

  String mongoJson = "{\"enum1\": \"X\", \"enum2\": \"A\", \"enum3\": null, \"enum4\": [\"SAT\", \"SUN\"]}}";
  BSONObject object = (BSONObject) JSON.parse(mongoJson);
  Record record2 = RecordConverter.toRecord(schema, object, getClass().getClassLoader());

  assertThat(record2, is(record1));
  assertThat(AvroHelper.toSimpleJson(schema, record2), is(AvroHelper.toSimpleJson(schema, record1)));
}

@Test
public void testMaps() throws Exception {
  Schema schema = Maps.SCHEMA$;

  GenericRecordBuilder builder = new GenericRecordBuilder(schema);
  builder.set("maps", ImmutableMap.of("key1", ImmutableMap.of("value1", 1, "value2", 2), "key2",
      ImmutableMap.of(), "key3", ImmutableMap.of("value3", 3)));
  Record record1 = builder.build();

  String json = "{\"maps\": {\"key1\": {\"value1\": 1, \"value2\": 2}, \"key2\": {}, \"key3\": {\"value3\": 3}}}";
  DBObject object = (DBObject) JSON.parse(json);
  Record record2 = RecordConverter.toRecord(schema, object, getClass().getClassLoader());

  // Convert into JsonNode before comparison, so the maps equal even if keys are reordered.
  assertThat(JSON.parse(AvroHelper.toSimpleJson(schema, record2)), is(JSON.parse(AvroHelper.toSimpleJson(schema, record1))));
}

@Test
public void testUnions() throws Exception {
  Schema schema = Unions.SCHEMA$;

  String avroJson = "{\"union1\": {\"int\": 1}, \"union2\": {\"test.Union2\": {\"union21\": {\"long\": 2}}}, \"union3\": {\"array\": [{\"boolean\": true}, {\"boolean\": false}, {\"null\": null}]}, \"union4\": {\"map\": {\"a\": {\"string\": \"A\"}, \"b\": {\"string\": \"B\"}, \"c\": {\"string\": \"C\"}}}, \"union5\": {\"null\": null}, \"union6\": {\"null\": null}}";
  Decoder decoder = DecoderFactory.get().jsonDecoder(schema, avroJson);
  GenericDatumReader<Record> reader = new GenericDatumReader<Record>(schema);
  Record record1 = reader.read(null, decoder);

  String mongoJson = "{\"union1\": 1, \"union2\": {\"union21\": 2}, \"union3\": [true, false, null], \"union4\": {\"a\": \"A\", \"b\": \"B\", \"c\": \"C\"}, \"union5\": null, \"union6\": null}";
  DBObject object = (DBObject) JSON.parse(mongoJson);
  Record record2 = RecordConverter.toRecord(schema, object, getClass().getClassLoader());

  assertThat(record2, is(record1));
  assertThat(AvroHelper.toSimpleJson(schema, record2), is(AvroHelper.toSimpleJson(schema, record1)));
}

@Override
public boolean start() throws IOException {
    super.start();

    String localFilePath = properties.localFilePath.getValue();
    String destinationFolderId = properties.destinationFolderAccessMethod.getValue().equals(AccessMethod.Id)
            ? properties.destinationFolder.getValue()
            : utils.getFolderId(properties.destinationFolder.getValue(), false);

    GoogleDrivePutParameters p = new GoogleDrivePutParameters(destinationFolderId, properties.fileName.getValue(),
            properties.overwrite.getValue(), localFilePath);
    sentFile = utils.putResource(p);
    record = new Record(properties.schemaMain.schema.getValue());
    record.put(0, java.nio.file.Files.readAllBytes(Paths.get(localFilePath)));
    record.put(1, sentFile.getParents().get(0));
    record.put(2, sentFile.getId());
    result.totalCount++;
    result.successCount++;

    return true;
}

@Override
public boolean start() throws IOException {
    super.start();
    //
    String resourceId = properties.fileAccessMethod.getValue().equals(AccessMethod.Id) ? properties.file.getValue()
            : utils.getFileId(properties.file.getValue());
    Map<String, MimeType> mimes = GoogleDriveMimeTypes.newDefaultMimeTypesSupported();
    mimes.put(MIME_TYPE_GOOGLE_DOCUMENT, properties.exportDocument.getValue());
    mimes.put(MIME_TYPE_GOOGLE_DRAWING, properties.exportDrawing.getValue());
    mimes.put(MIME_TYPE_GOOGLE_PRESENTATION, properties.exportPresentation.getValue());
    mimes.put(MIME_TYPE_GOOGLE_SPREADSHEET, properties.exportSpreadsheet.getValue());
    GoogleDriveGetParameters p = new GoogleDriveGetParameters(resourceId, mimes, properties.storeToLocal.getValue(),
            properties.outputFileName.getValue(), properties.setOutputExt.getValue());
    //
    GoogleDriveGetResult r = utils.getResource(p);
    fileId = r.getId();
    byte[] content = r.getContent();
    record = new Record(properties.schemaMain.schema.getValue());
    record.put(0, content);
    result.totalCount++;
    result.successCount++;

    return true;
}

public List<IndexedRecord> convertLeadRecords(List<Map<String, String>> records, Schema schema,
        Map<String, String> mappings) {
    List<IndexedRecord> results = new ArrayList<>();
    for (Map<String, String> input : records) {
        IndexedRecord record = new Record(schema);
        for (Field f : schema.getFields()) {
            String col = mappings.get(f.name());
            Object tmp = input.get(col);
            if (col != null) {
                record.put(f.pos(), getValueType(f, tmp));
            }
        }
        results.add(record);
    }
    return results;
}

@Test
public void testWriteSOAP() throws Exception {
    doReturn(getSuccessSyncResult("added")).when(client).addToList(any(ListOperationParameters.class));
    IndexedRecord record = new Record(MarketoConstants.getListOperationSOAPSchema());
    record.put(0, "MKTOLISTNAME");
    record.put(1, "TESTS");
    record.put(2, "ID");
    record.put(3, "12345");
    props.connection.apiMode.setValue(APIMode.SOAP);
    props.dieOnError.setValue(false);
    props.multipleOperation.setValue(true);
    props.schemaInput.schema.setValue(MarketoConstants.getListOperationSOAPSchema());
    props.updateOutputSchemas();
    when(client.getApi()).thenReturn("SOAP");
    when(sink.getProperties()).thenReturn(props);
    writer.open("test");
    writer.write(record);
    assertEquals(1, writer.getSuccessfulWrites().size());
    record.put(1, "TEST2");
    writer.write(record);
    assertNotNull(writer.close());
    assertEquals(1, writer.getSuccessfulWrites().size());
    //
}

@Test
public void testRetryOperationFailDieOnError() throws Exception {
    IndexedRecord record = new Record(MarketoConstants.getListOperationRESTSchema());
    record.put(0, 12345);
    record.put(1, 54321);
    doReturn(false).when(client).isErrorRecoverable(any(List.class));
    doReturn(getFailedSyncResult("REST", "902", "Invalid operation")).when(client)
            .addToList(any(ListOperationParameters.class));
    writer.open("test");
    writer.write(record);
    try {
        writer.close();
        fail("Should not be here");
    } catch (Exception e) {
        assertTrue(e.getMessage().contains("902"));
    }
}

@Test
public void testRetryOperationFailNonRecoverableErrror() throws Exception {
    IndexedRecord record = new Record(MarketoConstants.getListOperationRESTSchema());
    record.put(0, 12345);
    record.put(1, 54321);
    doReturn(false).when(client).isErrorRecoverable(any(List.class));
    doReturn(getFailedSyncResult("REST", "902", "Invalid operation")).when(client)
            .addToList(any(ListOperationParameters.class));
    props.dieOnError.setValue(false);
    when(sink.getProperties()).thenReturn(props);
    writer.open("test");
    writer.write(record);
    MarketoResult result = (MarketoResult) writer.close();
    assertEquals(1, result.apiCalls);
    assertEquals(Collections.emptyList(), writer.getSuccessfulWrites());
    List<IndexedRecord> rejects = writer.getRejectedWrites();
    IndexedRecord reject = rejects.get(0);
    assertNotNull(reject);
    assertEquals("failed", reject.get(2));
    assertTrue(String.valueOf(reject.get(3)).contains("902"));
}

@Test
public void testRetryOperationFailRecoverableErrror() throws Exception {
    IndexedRecord record = new Record(MarketoConstants.getListOperationRESTSchema());
    record.put(0, 12345);
    record.put(1, 54321);
    doReturn(getFailedSyncResult("REST", "602", "expired header")).when(client).addToList(any(ListOperationParameters.class));
    doReturn(true).when(client).isErrorRecoverable(any(List.class));
    props.dieOnError.setValue(false);
    when(sink.getProperties()).thenReturn(props);
    int minDelay = props.connection.maxReconnAttemps.getValue() * props.connection.attemptsIntervalTime.getValue();
    long start = System.currentTimeMillis();
    writer.open("test");
    writer.write(record);
    MarketoResult result = (MarketoResult) writer.close();
    long end = System.currentTimeMillis();
    assertEquals((long) props.connection.maxReconnAttemps.getValue(), result.apiCalls);
    assertEquals(Collections.emptyList(), writer.getSuccessfulWrites());
    assertTrue(minDelay <= (end - start));
    List<IndexedRecord> rejects = writer.getRejectedWrites();
    IndexedRecord reject = rejects.get(0);
    assertNotNull(reject);
    assertEquals("failed", reject.get(2));
    assertTrue(String.valueOf(reject.get(3)).contains("602"));
}

@Test
public void testWriteDeleteLeads() throws Exception {
    writer.open("test");
    writer.write(null);
    // deleteLeads
    doReturn(getFailedSyncResult(true)).when(client).deleteLeads(any(ArrayList.class));
    IndexedRecord record = new Record(MarketoConstants.getDeleteLeadsSchema());
    record.put(0, 12345);
    try {
        writer.write(record);
        fail("Should not be here");
    } catch (Exception e) {
    }
    props.dieOnError.setValue(false);
    when(sink.getProperties()).thenReturn(props);
    writer.open("test");
    writer.write(record);
    assertNotNull(writer.close());
}

@Test
public void testWriteCustomObject() throws Exception {
    props.outputOperation.setValue(OutputOperation.deleteCustomObjects);
    props.updateSchemaRelated();
    when(sink.getProperties()).thenReturn(props);
    doReturn(getSuccessSyncResult("deleted")).when(client).deleteCustomObjects(any(TMarketoOutputProperties.class),
            any(List.class));
    IndexedRecord record = new Record(MarketoConstants.getRESTOutputSchemaForSyncLead());
    record.put(0, 12345);
    writer.open("test");
    writer.write(record);
    assertNotNull(writer.close());
    //
    props.outputOperation.setValue(OutputOperation.syncCustomObjects);
    when(sink.getProperties()).thenReturn(props);
    doReturn(getSuccessSyncResult("updated")).when(client).syncCustomObjects(any(TMarketoOutputProperties.class),
            any(List.class));
    writer.open("test");
    writer.write(record);
    assertNotNull(writer.close());
}

@Test
public void testRetryOperationFailDieOnError() throws Exception {
    IndexedRecord record = new Record(MarketoConstants.getRESTOutputSchemaForSyncMultipleLeads());
    record.put(0, 12345);
    doReturn(false).when(client).isErrorRecoverable(any(List.class));
    doReturn(getFailedSyncResult("REST", "902", "Invalid operation")).when(client)
            .syncMultipleLeads(any(TMarketoOutputProperties.class), any(List.class));
    props.outputOperation.setValue(OutputOperation.syncMultipleLeads);
    props.updateSchemaRelated();
    when(sink.getProperties()).thenReturn(props);
    writer.open("test");
    try {
        writer.write(record);
        writer.close();
        fail("Should not be here");
    } catch (Exception e) {
        assertTrue(e.getMessage().contains("902"));
    }
}

@Test
public void testRetryOperationFailNonRecoverableErrror() throws Exception {
    IndexedRecord record = new Record(MarketoConstants.getRESTOutputSchemaForSyncMultipleLeads());
    record.put(0, 12345);
    doReturn(false).when(client).isErrorRecoverable(any(List.class));
    doReturn(getFailedSyncResult("REST", "902", "Invalid operation")).when(client)
            .syncMultipleLeads(any(TMarketoOutputProperties.class), any(List.class));
    props.dieOnError.setValue(false);
    props.outputOperation.setValue(OutputOperation.syncMultipleLeads);
    props.updateSchemaRelated();
    when(sink.getProperties()).thenReturn(props);
    writer.open("test");
    writer.write(record);
    MarketoResult result = (MarketoResult) writer.close();
    assertEquals(1, result.apiCalls);
    assertEquals(Collections.emptyList(), writer.getSuccessfulWrites());
    List<IndexedRecord> rejects = writer.getRejectedWrites();
    IndexedRecord reject = rejects.get(0);
    assertNotNull(reject);
    assertEquals("failed", reject.get(4));
    assertTrue(String.valueOf(reject.get(5)).contains("902"));
}

@Test
public void testRetryOperationFailRecoverableErrror() throws Exception {
    IndexedRecord record = new Record(MarketoConstants.getRESTOutputSchemaForSyncMultipleLeads());
    record.put(0, 12345);
    doReturn(getFailedSyncResult("REST", "602", "expired header")).when(client)
            .syncMultipleLeads(any(TMarketoOutputProperties.class), any(List.class));
    doReturn(true).when(client).isErrorRecoverable(any(List.class));
    props.dieOnError.setValue(false);
    props.outputOperation.setValue(OutputOperation.syncMultipleLeads);
    props.updateSchemaRelated();
    when(sink.getProperties()).thenReturn(props);
    int minDelay = props.connection.maxReconnAttemps.getValue() * props.connection.attemptsIntervalTime.getValue();
    long start = System.currentTimeMillis();
    writer.open("test");
    writer.write(record);
    MarketoResult result = (MarketoResult) writer.close();
    long end = System.currentTimeMillis();
    assertEquals((long) props.connection.maxReconnAttemps.getValue(), result.apiCalls);
    assertEquals(Collections.emptyList(), writer.getSuccessfulWrites());
    assertTrue(minDelay <= (end - start));
    List<IndexedRecord> rejects = writer.getRejectedWrites();
    IndexedRecord reject = rejects.get(0);
    assertNotNull(reject);
    assertEquals("failed", reject.get(4));
    assertTrue(String.valueOf(reject.get(5)).contains("602"));
}

public MarketoRecordResult getCustomObjectResult() {
    MarketoRecordResult cor = new MarketoRecordResult();
    cor.setSuccess(true);
    List<IndexedRecord> cos = new ArrayList<>();
    IndexedRecord co = new Record(MarketoConstants.getCustomObjectDescribeSchema());
    co.put(0, "car_c");
    co.put(1, "marketoGUID");
    co.put(2, "Car");
    co.put(3, "Car system");
    co.put(4, new Date());
    co.put(5, new Date());
    co.put(6, "");
    co.put(7, "{ \"brand\", \"model\" }");
    co.put(8, "{}");
    co.put(9, "{}");
    cos.add(co);
    cor.setRecords(cos);
    return cor;
}

@Test
public void testSyncCustomObjects() throws Exception {
    oprops.customObjectSyncAction.setValue(CustomObjectSyncAction.createOrUpdate);
    oprops.customObjectDedupeBy.setValue("marketoGUID");
    //
    doThrow(new MarketoException("REST", "error")).when(client).executePostRequest(eq(SyncResult.class),
            any(JsonObject.class));
    List<IndexedRecord> records = new ArrayList<>();
    IndexedRecord record = new Record(MarketoConstants.getCustomObjectRecordSchema());
    record.put(0, "mkto-123456");
    records.add(record);
    mktoSR = client.syncCustomObjects(oprops, records);
    assertFalse(mktoSR.isSuccess());
    assertFalse(mktoSR.getErrorsString().isEmpty());
    //
    doReturn(new SyncResult()).when(client).executePostRequest(eq(SyncResult.class), any(JsonObject.class));
    mktoSR = client.syncCustomObjects(oprops, records);
    assertFalse(mktoSR.isSuccess());
    //
    doReturn(getListOperationResult(true, "deleted")).when(client).executePostRequest(eq(SyncResult.class),
            any(JsonObject.class));
    mktoSR = client.syncCustomObjects(oprops, records);
    assertTrue(mktoSR.isSuccess());
    assertTrue(mktoSR.getErrorsString().isEmpty());
}

@Test
public void testDeleteCustomObjects() throws Exception {
    oprops.customObjectDeleteBy.setValue(CustomObjectDeleteBy.idField);
    //
    doThrow(new MarketoException("REST", "error")).when(client).executePostRequest(eq(SyncResult.class),
            any(JsonObject.class));
    List<IndexedRecord> records = new ArrayList<>();
    IndexedRecord record = new Record(MarketoConstants.getCustomObjectRecordSchema());
    record.put(0, "mkto-123456");
    records.add(record);
    mktoSR = client.deleteCustomObjects(oprops, records);
    assertFalse(mktoSR.isSuccess());
    assertFalse(mktoSR.getErrorsString().isEmpty());
    //
    doReturn(new SyncResult()).when(client).executePostRequest(eq(SyncResult.class), any(JsonObject.class));
    mktoSR = client.deleteCustomObjects(oprops, records);
    assertFalse(mktoSR.isSuccess());
    //
    doReturn(getListOperationResult(true, "deleted")).when(client).executePostRequest(eq(SyncResult.class),
            any(JsonObject.class));
    mktoSR = client.deleteCustomObjects(oprops, records);
    assertTrue(mktoSR.isSuccess());
    assertTrue(mktoSR.getErrorsString().isEmpty());
}

public AvroGenericRecordWritable decode(byte[] payload) {
    try {
        MessageDecoderHelper helper = new MessageDecoderHelper(registry,
                topicName, payload).invoke();
     DatumReader<Record> reader = new GenericDatumReader<Record>(helper.getTargetSchema());

     log.debug("Trying to read kafka payload");
     log.debug("buffer: " + helper.getBuffer());
     log.debug("start: " + helper.getStart());
     log.debug("length: " + helper.getLength());
     log.debug("target schema: " + helper.getTargetSchema());
     log.debug("schema: " + helper.getSchema());
      GenericRecord record = reader.read(null, decoderFactory.binaryDecoder(helper.getBuffer().array(),
         helper.getStart(), helper.getLength(), null));
     log.debug("Read kafka payload as " + record);

     AvroGenericRecordWritable grw = new AvroGenericRecordWritable(record);
     grw.setFileSchema(latestSchema);

     return grw;

    } catch (IOException e) {
        throw new MessageDecoderException(e);
    }
}

private void initCurrentFile() throws IOException {
  if (reader != null) {
    reader.close();
  }

  LOG.info("Initializing {}:{}+{}",
      new Object[] { split.getPath(currentFile),
          split.getOffset(currentFile), split.getLength(currentFile) });

  GenericDatumReader<Record> datumReader = new GenericDatumReader<Record>(
      Schemas.getSchema("logBlock"));
  reader = new DataFileReader<Record>(new FsInput(split.getPath(currentFile),
      job), datumReader);
  datumReader.setExpected(Schemas.getSchema("logBlock"));
  datumReader.setSchema(reader.getSchema());

  long size = split.getLength(currentFile);
  start = split.getOffset(currentFile);
  end = start + size;

  reader.sync(start);
}

/**
 * Matches type.
 *
 * @param type the type
 * @param jsonValue the json value
 * @return true, if successful
 */
private static boolean matchesType(Record type, JsonNode jsonValue) {
    if (type.getSchema().getName().equals(BOOLEAN_FIELD_TYPE)) {
        return jsonValue.isBoolean();
    } else if (type.getSchema().getName().equals(INTEGER_FIELD_TYPE)) {
        return jsonValue.isInt();
    } else if (type.getSchema().getName().equals(LONG_FIELD_TYPE)) {
        return jsonValue.isIntegralNumber();
    } else if (type.getSchema().getName().equals(FLOAT_FIELD_TYPE)) {
        return jsonValue.isDouble();
    } else if (type.getSchema().getName().equals(DOUBLE_FIELD_TYPE)) {
        return jsonValue.isFloatingPointNumber();
    } else if (type.getSchema().getName().equals(STRING_FIELD_TYPE) || 
            type.getSchema().getName().equals(ENUM_FIELD_TYPE)) {
        return jsonValue.isTextual();
    } else if (type.getSchema().getName().equals(BYTES_FIELD_TYPE) || 
            type.getSchema().getName().equals(FIXED_FIELD_TYPE)) {
        return jsonValue.isBinary() || jsonValue.isArray();
    } else {
        return false;
    }
}

@Test(dataProvider = "google-ips")
public void testIspEventMapper(Map<String, Object> props, InetAddress address)
        throws Exception {
    MaxmindGeoIPEventMapper mapper = new MaxmindGeoIPEventMapper(new MaxmindGeoIPModuleConfig()
            .setAttributes("")
            .setIspDatabaseUrl(new URL("https://github.com/maxmind/MaxMind-DB/raw/master/test-data/GeoIP2-ISP-Test.mmdb")));
    FieldDependencyBuilder builder = new FieldDependencyBuilder();
    mapper.addFieldDependency(builder);

    Record properties = new Record(Schema.createRecord(ImmutableList.of(
            new Schema.Field("_ip", Schema.create(NULL), null, null),
            new Schema.Field("__ip", Schema.create(STRING), null, null),
            new Schema.Field("_isp", Schema.create(STRING), null, null))));
    props.forEach(properties::put);

    Event event = new Event("testproject", "testcollection", null, null, properties);

    List<Cookie> resp = mapper.map(event, EventMapper.RequestParams.EMPTY_PARAMS, address, null);

    assertTrue(resp == null);

    assertEquals(event.getAttribute("_isp"), "Level 3 Communications");
    GenericData.get().validate(properties.getSchema(), properties);
}

@Test(dataProvider = "google-ips")
public void testConnectionTypeEventMapper(Map<String, Object> props, InetAddress address)
        throws Exception {
    MaxmindGeoIPEventMapper mapper = new MaxmindGeoIPEventMapper(new MaxmindGeoIPModuleConfig()
            .setAttributes("")
            .setConnectionTypeDatabaseUrl(new URL("https://github.com/maxmind/MaxMind-DB/raw/master/test-data/GeoIP2-Connection-Type-Test.mmdb")));
    FieldDependencyBuilder builder = new FieldDependencyBuilder();
    mapper.addFieldDependency(builder);

    Record properties = new Record(Schema.createRecord(ImmutableList.of(
            new Schema.Field("_ip", Schema.create(NULL), null, null),
            new Schema.Field("__ip", Schema.create(STRING), null, null),
            new Schema.Field("_connection_type", Schema.create(STRING), null, null))));
    props.forEach(properties::put);

    Event event = new Event("testproject", "testcollection", null, null, properties);

    List<Cookie> resp = mapper.map(event, EventMapper.RequestParams.EMPTY_PARAMS, address, null);

    assertTrue(resp == null);

    // TODO: find a reliable ip that can be mapped.
    assertNull(event.getAttribute("connection_type"));
    GenericData.get().validate(properties.getSchema(), properties);
}

protected void writeLine(Context context, byte[] rowKey) throws IOException, InterruptedException {

      if (columnValueMap.size() > 0) {
        Record record = new Record(schema);

        for (String col : columns) {


          byte[] value = columnValueMap.get(col);

          System.out.print(col + "," + Bytes.toString(value));

          if (value != null) {
            putValue(record, col, value);
          } else if (col.equals(rowKeyColumn)) {
            putValue(record, col, rowKey);
          }
        }
        System.out.println();
        context.write(null, record);
      }
    }

@Override
protected void customizeFormField(Record fieldType, Field field) {
  if (fieldType.getSchema().getName().equals(ARRAY_FIELD_TYPE)) {
    JsonNode overrideStrategyNode = field.getJsonProp(OVERRIDE_STRATEGY);
    Schema overrideStrategySchema =
        fieldType.getSchema().getField(OVERRIDE_STRATEGY).schema();
    if (overrideStrategyNode != null && overrideStrategyNode.isTextual()) {
      fieldType.put(OVERRIDE_STRATEGY,
          new GenericData.EnumSymbol(
              overrideStrategySchema, overrideStrategyNode.asText().toUpperCase()));
    } else {
      fieldType.put(OVERRIDE_STRATEGY,
          new GenericData.EnumSymbol(
              overrideStrategySchema, OverrideStrategy.REPLACE.name()));
    }
  }
}

@Override
public void open(Configuration conf,
                 JsonNode json,
                 BlockSchema schema,
                 Path root,
                 String filename) throws IOException
{
    Path teePath = new Path(root, filename + ".avro");
    FileSystem fs = FileSystem.get(conf);

    Schema avroSchema = AvroUtils.convertFromBlockSchema("record", schema);
    record = new Record(avroSchema);
    numColumns = schema.getNumColumns();

    DatumWriter<GenericRecord> datumWriter =
            new GenericDatumWriter<GenericRecord>(avroSchema);
    dataFileWriter = new DataFileWriter<GenericRecord>(datumWriter);
    dataFileWriter.create(avroSchema, fs.create(teePath));
}