Java 类org.apache.avro.generic.IndexedRecord 实例源码

private void recurse(String context, TypeDescriptor<?> type, Schema schema) {
  if (type.getRawType().isAnnotationPresent(AvroSchema.class)) {
    reportError(context, "Custom schemas are not supported -- remove @AvroSchema.");
    return;
  }

  if (!activeTypes.add(type)) {
    reportError(context, "%s appears recursively", type);
    return;
  }

  // If the the record isn't a true class, but rather a GenericRecord, SpecificRecord, etc.
  // with a specified schema, then we need to make the decision based on the generated
  // implementations.
  if (isSubtypeOf(type, IndexedRecord.class)) {
    checkIndexedRecord(context, schema, null);
  } else {
    doCheck(context, type, schema);
  }

  activeTypes.remove(type);
}

private void checkRecord(TypeDescriptor<?> type, Schema schema) {
  // For a record, we want to make sure that all the fields are deterministic.
  Class<?> clazz = type.getRawType();
  for (org.apache.avro.Schema.Field fieldSchema : schema.getFields()) {
    Field field = getField(clazz, fieldSchema.name());
    String fieldContext = field.getDeclaringClass().getName() + "#" + field.getName();

    if (field.isAnnotationPresent(AvroEncode.class)) {
      reportError(fieldContext,
          "Custom encoders may be non-deterministic -- remove @AvroEncode");
      continue;
    }

    if (!IndexedRecord.class.isAssignableFrom(field.getType())
        && field.isAnnotationPresent(AvroSchema.class)) {
      // TODO: We should be able to support custom schemas on POJO fields, but we shouldn't
      // need to, so we just allow it in the case of IndexedRecords.
      reportError(fieldContext,
          "Custom schemas are only supported for subtypes of IndexedRecord.");
      continue;
    }

    TypeDescriptor<?> fieldType = type.resolveType(field.getGenericType());
    recurse(fieldContext, fieldType, fieldSchema.schema());
  }
}

@Test
public void testHierarchical_TFD2119_ERR4_NullArray() throws Exception {
    DoFnTester<Object, IndexedRecord> fnTester = DoFnTester.of( //
            new FilterRowDoFn().withProperties(addCriteria(null, //
                    ".b1[0].id", //
                    null, //
                    ConditionsRowConstant.Operator.EQUAL, //
                    "1") //

            ).withOutputSchema(true));

    // Looks like this is not an exception -- it just doesn't match.
    IndexedRecord[] input = copyAndReplaceSubrecordArray(inputB, 10, true);
    List<IndexedRecord> output = fnTester.processBundle(input);
    for (IndexedRecord main : output) {
        List<IndexedRecord> subrecords = getSubrecords(main);
        assertThat(main.toString(), subrecords.get(0).get(0), is((Object) 1));
    }
    assertThat(output, hasSize(102));
}

@SuppressWarnings("rawtypes")
@Test
public void testListQueues() throws Throwable {
    TAzureStorageQueueListProperties properties = new TAzureStorageQueueListProperties("tests");
    properties.setupProperties();
    properties = (TAzureStorageQueueListProperties) setupConnectionProperties(
            (AzureStorageProvideConnectionProperties) properties);
    BoundedReader reader = createBoundedReader(properties);
    assertTrue(reader.start());
    do {
        IndexedRecord current = (IndexedRecord) reader.getCurrent();
        assertNotNull(current);
        assertTrue(current.get(0) instanceof String);
    } while (reader.advance());
    assertTrue((int) reader.getReturnValues().get(AzureStorageDefinition.RETURN_TOTAL_RECORD_COUNT) > 0);
    reader.close();
}

@SuppressWarnings("rawtypes")
@Test
public void testSystemReader() throws Throwable {
    String ctable = tbl_test + "InputSys";
    createSampleDataset(ctable);

    properties.tableName.setValue(ctable);
    properties.schema.schema.setValue(getSystemSchema());
    properties.useFilterExpression.setValue(false);
    BoundedReader reader = createBoundedReader(properties);
    assertTrue(reader.start());
    while (reader.advance()) {
        IndexedRecord current = (IndexedRecord) reader.getCurrent();
        assertNotNull(current);
        assertEquals(getSystemSchema(), current.getSchema());
        assertEquals(3, current.getSchema().getFields().size());
    }
    reader.close();
}

/**
 * Test aggregate query field not case sensitive
 */
@Test
public void testAggregateQueryColumnNameCaseSensitive() throws Throwable {
    TSalesforceInputProperties props = createTSalesforceInputProperties(true, false);
    props.manualQuery.setValue(true);
    props.query.setValue("SELECT MIN(CreatedDate) value FROM Contact GROUP BY FirstName, LastName LIMIT 1");
    props.module.main.schema.setValue(SCHEMA_DATE);
    List<IndexedRecord> outputRows = readRows(props);
    if (outputRows.isEmpty()) {
        return;
    }
    IndexedRecord record = outputRows.get(0);
    assertNotNull(record.getSchema());
    Object value = record.get(0);
    Assert.assertTrue(value != null && value instanceof Long);
}

@Test
public void test_OneOutputRow() throws Exception {
    String inputAsString = generateInputJSON(inputSchema, inputIndexedRecord1);

    FixedFlowInputProperties properties = new FixedFlowInputProperties("test");
    properties.init();
    properties.schemaFlow.schema.setValue(inputSchema);
    properties.values.setValue(inputAsString);
    properties.nbRows.setValue(1);

    FixedFlowInputRuntime runtime = new FixedFlowInputRuntime();
    runtime.initialize(null, properties);

    PCollection<IndexedRecord> indexRecords = pipeline.apply(runtime);
    try (DirectCollector<IndexedRecord> collector = DirectCollector.of()) {
        indexRecords.apply(collector);

        // Run the pipeline to fill the collectors.
        pipeline.run().waitUntilFinish();;

        // Validate the contents of the collected outputs.
        List<IndexedRecord> outputs = collector.getRecords();
        assertEquals(1, outputs.size());
        assertEquals(inputIndexedRecord1.toString(), outputs.get(0).toString());
    }
}

public void createSampleDataset(String table) throws Throwable {
    tableClient.getTableReference(table).createIfNotExists();
    TAzureStorageOutputTableProperties props = new TAzureStorageOutputTableProperties("tests");
    props = (TAzureStorageOutputTableProperties) setupConnectionProperties(props);
    props.setupProperties();
    props.schema.schema.setValue(getDynamicSchema());
    props.actionOnTable.setValue(ActionOnTable.Default);
    props.actionOnData.setValue(ActionOnData.Insert);
    props.schemaListener.afterSchema();
    props.tableName.setValue(table);
    Writer<?> writer = createWriter(props);
    writer.open("test-uid");
    for (String p : partitions) {
        for (String r : rows) {
            IndexedRecord entity = new GenericData.Record(getWriteSchema());
            entity.put(0, p);
            entity.put(1, r);
            entity.put(2, RandomStringUtils.random(50));
            entity.put(3, RandomStringUtils.randomNumeric(10));
            writer.write(entity);
        }
    }
    writer.close();
}

@Test
public void testHierarchical_TFD2119_B1_AtLeastOneSubRecordHasValueGt10() throws Exception {
    DoFnTester<Object, IndexedRecord> fnTester = DoFnTester.of( //
            new FilterRowDoFn().withProperties(addCriteria(null, //
                    ".b1{.value > 10}", //
                    ConditionsRowConstant.Function.COUNT, //
                    ConditionsRowConstant.Operator.GREATER, //
                    "0") //

            ).withOutputSchema(true));

    List<IndexedRecord> output = fnTester.processBundle(inputB);
    for (IndexedRecord main : output) {
        boolean atLeastOne = false;
        for (IndexedRecord subrecord : getSubrecords(main)) {
            if ((double) subrecord.get(2) > 10)
                atLeastOne = true;
        }
        assertThat(main.toString(), atLeastOne, is(true));
    }
    assertThat(output, hasSize(274));
}

private void getSampleAction(JDBCDatasetProperties dataset) {
    JDBCDatasetRuntime runtime = new JDBCDatasetRuntime();
    runtime.initialize(null, dataset);
    final IndexedRecord[] record = new IndexedRecord[1];
    Consumer<IndexedRecord> storeTheRecords = new Consumer<IndexedRecord>() {

        @Override
        public void accept(IndexedRecord data) {
            record[0] = data;

        }
    };

    runtime.getSample(1, storeTheRecords);
    Assert.assertEquals("1", record[0].get(0));
    Assert.assertEquals("wangwei", record[0].get(1));
}

@Test
public void testGetCustomObjectWithCompoundKey() throws Exception {
    irProps.customObjectAction.setValue(CustomObjectAction.get);
    irProps.customObjectName.setValue(TEST_CO_NAME_CAR);
    irProps.validateFetchCustomObjectSchema();
    irProps.useCompoundKey.setValue(true);
    // "searchableFields": "[[\"customerId\",\"VIN\"],[\"marketoGUID\"],[\"customerId\"]]"
    irProps.compoundKey.keyName.setValue(Arrays.asList("customerId", "VIN"));
    irProps.compoundKey.keyValue.setValue(Arrays.asList("4137181", "WBA4R7C30HK896061"));// WBA4R7C55HK895912
    MarketoSource source = new MarketoSource();
    source.initialize(null, irProps);
    MarketoRESTClient client = (MarketoRESTClient) source.getClientService(null);
    MarketoRecordResult result = client.getCustomObjects(irProps, null);
    LOG.debug("result = {}.", result);
    assertNotNull(result.getRecords());
    assertEquals(1, result.getRecords().size());
    IndexedRecord record = result.getRecords().get(0);
    Schema s = record.getSchema();
    assertEquals(4137181, record.get(s.getField("customerId").pos()));
    assertEquals("WBA4R7C30HK896061", record.get(s.getField("VIN").pos()));
    assertEquals("FIT", record.get(s.getField("brand").pos()));
}

private void map(IndexedRecord input, ProcessContext context) throws IOException {
    // Prepare Python environment
    interpretor.set("inputJSON", new PyString(input.toString()));

    // Add user command
    interpretor.exec(pythonFunction);

    // Retrieve results
    interpretor.exec("outputJSON = userFunction(inputJSON)");
    PyObject output = interpretor.get("outputJSON");

    if (jsonGenericRecordConverter == null) {
        JsonSchemaInferrer jsonSchemaInferrer = new JsonSchemaInferrer(new ObjectMapper());
        Schema jsonSchema = jsonSchemaInferrer.inferSchema(output.toString());
        jsonGenericRecordConverter = new JsonGenericRecordConverter(jsonSchema);
    }

    GenericRecord outputRecord = jsonGenericRecordConverter.convertToAvro(output.toString());
    context.output(outputRecord);
}

@Test (expected = UnmodifiableAdapterException.class)
public void testConvertToAvroFailedToChangeRecord() {
    ByteBuf buffer = Mockito.mock(ByteBuf.class);

    // Mocking key object
    Mockito.when(buffer.getByte(0)).thenReturn(MessageUtil.MAGIC_REQ);
    Mockito.when(buffer.getByte(1)).thenReturn(MessageUtil.DCP_EXPIRATION_OPCODE);
    Mockito.when(buffer.getByte(4)).thenReturn(OFFSET);
    Mockito.when(buffer.getShort(2)).thenReturn(LENGTH);
    ByteBuf key = Mockito.mock(ByteBuf.class);
    Mockito.when(key.readableBytes()).thenReturn(4);
    Mockito.when(buffer.slice(29, 10)).thenReturn(key);

    IndexedRecord record = converter.convertToAvro(buffer);

    record.put(0, "Update is not supported");
}

@Test
public void testArrayOfArrayOfSimpleType() throws IOException {
    RecordingVisitor visitor = new RecordingVisitor();
    IndexedRecord record = loadRecord("arrayOfArrayOfSimpleType");

    VisitableRecord wrapper = new VisitableRecord(record);
    wrapper.accept(visitor);

    List<List<Integer>> array = createArrayOfArray();

    visitor.verifyRoot();
    visitor.verifyField("/intField", 123);
    visitor.verifyField("/arrayOfArrayOfSimpleType", array);
    visitor.verifyField("/arrayOfArrayOfSimpleType[0]", array.get(0));
    visitor.verifyField("/arrayOfArrayOfSimpleType[0][0]", array.get(0).get(0));
    visitor.verifyField("/arrayOfArrayOfSimpleType[0][1]", array.get(0).get(1));
    visitor.verifyField("/arrayOfArrayOfSimpleType[0][2]", array.get(0).get(2));
    visitor.verifyField("/arrayOfArrayOfSimpleType[1]", array.get(1));
    visitor.verifyField("/arrayOfArrayOfSimpleType[1][0]", array.get(1).get(0));
    visitor.verifyField("/arrayOfArrayOfSimpleType[1][1]", array.get(1).get(1));
    visitor.verifyField("/arrayOfArrayOfSimpleType[1][2]", array.get(1).get(2));
    Assert.assertTrue("Visitor not verified", visitor.isVerified());
}

@Test
public void testStart() throws Exception {
    ValidationResult vr = source.initialize(container, properties);
    assertNotNull(vr);
    assertEquals(Result.OK, vr.getStatus());
    BoundedReader reader = source.createReader(container);
    assertTrue(reader.start());
    IndexedRecord record = (IndexedRecord) reader.getCurrent();
    assertNotNull(record);
    assertEquals(1, record.getSchema().getFields().size());
    assertFalse(reader.advance());
    reader.close();
    Map<String, Object> returnValues = reader.getReturnValues();
    assertEquals(FILE_GET_ID, returnValues.get(GoogleDriveGetDefinition.RETURN_FILE_ID));
    assertNull(returnValues.get(GoogleDriveGetDefinition.RETURN_CONTENT));
}

@Test
public void testBasic() throws Exception {
    NetSuiteClientService<?> connection = webServiceTestFixture.getClientService();
    connection.login();

    TypeDesc typeDesc = connection.getMetaDataSource().getTypeInfo("Opportunity");
    Schema schema = NetSuiteDatasetRuntimeImpl.inferSchemaForType(typeDesc.getTypeName(), typeDesc.getFields());

    NsObjectInputTransducer transducer = new NsObjectInputTransducer(connection, schema, typeDesc.getTypeName());

    SearchResultSet<Record> rs = connection.newSearch()
            .target(typeDesc.getTypeName())
            .search();

    if (!rs.next()) {
        throw new IllegalStateException("Not records");
    }

    Record record = rs.get();

    IndexedRecord indexedRecord = transducer.read(record);
    logger.debug("Indexed record: {}", indexedRecord);
}

@SuppressWarnings({ "rawtypes" })
@Test
public void testType() throws Exception {
    JDBCInputDefinition definition = new JDBCInputDefinition();
    JDBCInputProperties properties = createCommonJDBCInputProperties(definition);

    Reader reader = DBTestUtils.createCommonJDBCInputReader(properties);

    try {
        IndexedRecordConverter<Object, ? extends IndexedRecord> converter = null;

        for (boolean available = reader.start(); available; available = reader.advance()) {
            converter = DBTestUtils.getIndexRecordConverter(reader, converter);

            IndexedRecord record = converter.convertToAvro(reader.getCurrent());

            assertEquals(String.class, record.get(0).getClass());
            assertEquals(String.class, record.get(1).getClass());
        }

        reader.close();
    } finally {
        reader.close();
    }
}

public void inputCsv(Pipeline pipeline, String streamName) throws IOException {
    String testID = "csvBasicTest" + new Random().nextInt();
    final String fieldDelimited = ";";

    List<Person> expectedPersons = Person.genRandomList(testID, maxRecords);
    List<IndexedRecord> expected = new ArrayList<>();
    KinesisInputRuntime.CsvConverter converter = new KinesisInputRuntime.CsvConverter(fieldDelimited);
    for (Person expectedPerson : expectedPersons) {
        String strPerson = expectedPerson.toCSV(fieldDelimited);
        amazonKinesis.putRecord(streamName, ByteBuffer.wrap(strPerson.getBytes("UTF-8")), expectedPerson.group);
        String[] data = strPerson.split(fieldDelimited);
        expected.add(new KinesisInputRuntime.StringArrayIndexedRecord(converter.inferStringArray(data), data));
    }

    runtime.initialize(null, getInputFromBeginning(getDatasetForCsv(getLocalDatastore(), streamName,
            KinesisDatasetProperties.FieldDelimiterType.SEMICOLON), null, maxRecords));

    PCollection<IndexedRecord> results = pipeline.apply(runtime);

    PAssert.that(results).containsInAnyOrder(expected);

    pipeline.run().waitUntilFinish();
}

private Consumer<IndexedRecord> getWritingConsumer(Encoder[] encoder) {
    return new Consumer<IndexedRecord>() {

        GenericDatumWriter<IndexedRecord> writer = null;

        @Override
        public void accept(IndexedRecord ir) {
            if (writer == null) {
                writer = new GenericDatumWriter<>(ir.getSchema());
                try {
                    if (json) {
                        encoder[0] = EncoderFactory.get().jsonEncoder(ir.getSchema(), output);
                    } else {
                        encoder[0] = EncoderFactory.get().binaryEncoder(output, null);
                    }
                } catch (IOException ioe) {
                    throw new RuntimeException(ioe);
                }

            }
            writeIndexedRecord(writer, encoder[0], ir);
        }
    };
}

@Test
public void testGetLeadsChanges() throws Exception {
    iprops.inputOperation.setValue(getLeadChanges);
    iprops.afterInputOperation();
    iprops.batchSize.setValue(100);
    iprops.sinceDateTime.setValue(DATE_OLDEST_CREATE);
    iprops.fieldList.setValue("id,email,firstName,lastName,company");
    iprops.beforeMappingInput();
    //
    MarketoSource source = new MarketoSource();
    source.initialize(null, iprops);
    MarketoClientService client = source.getClientService(null);
    //
    MarketoRecordResult result = client.getLeadChanges(iprops, null);
    LOG.debug("{}", result);
    List<IndexedRecord> changes = result.getRecords();
    assertTrue(changes.size() > 0);
    assertTrue(result.getRemainCount() > 0);
}

@Test
public void testStartCopyFile() throws Exception {
    source.initialize(container, properties);
    BoundedReader reader = source.createReader(container);
    assertTrue(reader.start());
    IndexedRecord record = (IndexedRecord) reader.getCurrent();
    assertNotNull(record);
    assertEquals(2, record.getSchema().getFields().size());
    assertEquals(SOURCE_ID, record.get(0));
    assertEquals(DESTINATION_ID, record.get(1));
    reader.close();
    Map<String, Object> returnValues = reader.getReturnValues();
    assertNotNull(returnValues);
    assertEquals(SOURCE_ID, returnValues.get(GoogleDriveCopyDefinition.RETURN_SOURCE_ID));
    assertEquals(DESTINATION_ID, returnValues.get(GoogleDriveCopyDefinition.RETURN_DESTINATION_ID));
}

private void handleReject(SQLException e) {
    Schema outSchema = CommonUtils.getRejectSchema((ComponentProperties) properties);
    IndexedRecord reject = new GenericData.Record(outSchema);

    for (Schema.Field outField : reject.getSchema().getFields()) {
        Object outValue = null;

        if ("errorCode".equals(outField.name())) {
            outValue = e.getSQLState();
        } else if ("errorMessage".equals(outField.name())) {
            outValue = e.getMessage();
        }

        reject.put(outField.pos(), outValue);
    }

    Map<String, Object> resultMessage = new HashMap<String, Object>();
    resultMessage.put("error", e.getMessage());
    resultMessage.put("talend_record", reject);
    throw new DataRejectException(resultMessage);
}

@BeforeClass
public static void setup() throws Throwable {
    randomizedValue = "Name_Unit_" + createNewRandom();

    List<IndexedRecord> outputRows = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
        GenericData.Record row = new GenericData.Record(getSchema(false));
        row.put("Name", randomizedValue);
        row.put("ShippingStreet", "123 Main Street");
        row.put("ShippingPostalCode", Integer.toString(i));
        row.put("BillingStreet", "123 Main Street");
        row.put("BillingState", "CA");
        row.put("BillingPostalCode", createNewRandom());
        outputRows.add(row);
    }

    writeRows(outputRows);
}

/**
 * Checks {@link JiraInsertWriter#write()} throws {@link IOException} which error message contains:
 * "Reason: record is invalid"
 * 
 * @throws IOException
 */
@Test
public void testWriteErrorMessage() throws IOException {
    IndexedRecord badIssueTypeRecord = new GenericData.Record(INSERT_SCHEMA);
    String insertIssue1 = "{\"fields\":{\"project\":{\"key\":\"TP\"},\"summary\":\"Integration test issue 1\",\"issuetype\":{\"id\":\"12345\"}}}";
    badIssueTypeRecord.put(0, insertIssue1);

    thrown.expect(IOException.class);
    thrown.expectMessage("Reason: record is invalid");
    thrown.expectMessage("Record: " + insertIssue1);
    thrown.expectMessage("Error: ");
    thrown.expectMessage("{\"errorMessages\":[],\"errors\":{\"issuetype\":\"valid issue type is required\"}}");

    JiraWriter insertIssueWriter = JiraTestsHelper.createWriter(HOST_PORT, USER, PASS, Resource.ISSUE, Action.INSERT);

    insertIssueWriter.open("ins");
    try {
        insertIssueWriter.write(badIssueTypeRecord);
    } finally {
        insertIssueWriter.close();
    }
}

@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 static <T> List<IndexedRecord> makeIndexedRecords(NetSuiteClientService<?> clientService, Schema schema,
        ObjectComposer<T> objectComposer, int count) throws Exception {

    NsObjectInputTransducer transducer = new NsObjectInputTransducer(clientService, schema, schema.getName());

    List<IndexedRecord> recordList = new ArrayList<>();

    while (count > 0) {
        T nsRecord = objectComposer.composeObject();

        IndexedRecord convertedRecord = transducer.read(nsRecord);
        Schema recordSchema = convertedRecord.getSchema();

        GenericRecord record = new GenericData.Record(recordSchema);
        for (Schema.Field field : schema.getFields()) {
            Object value = convertedRecord.get(field.pos());
            record.put(field.pos(), value);
        }

        recordList.add(record);

        count--;
    }

    return recordList;
}

private byte[] encodeAvroGenericRecord(Schema schema, GenericRecord record) throws IOException {
  DatumWriter<IndexedRecord> msgDatumWriter = new GenericDatumWriter<>(schema);
  ByteArrayOutputStream os = new ByteArrayOutputStream();
  Encoder encoder = EncoderFactory.get().binaryEncoder(os, null);
  msgDatumWriter.write(record, encoder);
  encoder.flush();
  return os.toByteArray();
}

private KineticaBulkInserter() {    
    try {
        gpudb = GPUConnectionManager.getInstance().getGPUdb();
        BulkInserter.WorkerList workers = getWorkers(table);

        bulkInserter = new BulkInserter
            <IndexedRecord>(gpudb, table.getName(), table.getType(), 
                KineticaConfiguration.BULKINGESTOR_BATCH_SIZE, getInsertUpdateOptions(), workers);
    } catch (Exception e) {
        throw new KineticaException(e);
    }
}

private void process(@Nonnull ConsumerRecords<String, IndexedRecord> records) {
  for (ConsumerRecord<String, IndexedRecord> record : records) {
    _receivedRecordCount++;
    _processor.process(record.value());
    if (_receivedRecordCount % 1000 == 0) {
      log.info("{}: received {} messages", _topic, _receivedRecordCount);
    }
  }
  _consumer.commitAsync();
}

private IndexedRecordConverter<IndexedRecord, IndexedRecord> getFactory(Object datum) {
    if (factory == null) {
        factory = new FileDelimitedIndexedRecordConverter();
        factory.setSchema(recordSchema);
        ((FileDelimitedIndexedRecordConverter) factory).setProperties(props);
    }
    return factory;
}

@Test
public void testBasicCsvCustomDelimiters() throws IOException, URISyntaxException {

    String inputFile = writeRandomCsvFile(mini.getFs(), "/user/test/input.csv", 0, 0, 10, 10, 6, "|", "---");
    String fileSpec = mini.getFs().getUri().resolve("/user/test/input.csv").toString();

    // Configure the component.
    SimpleFileIOInputProperties inputProps = createInputComponentProperties();
    inputProps.getDatasetProperties().path.setValue(fileSpec);
    inputProps.getDatasetProperties().recordDelimiter.setValue(RecordDelimiterType.OTHER);
    inputProps.getDatasetProperties().specificRecordDelimiter.setValue("---");
    inputProps.getDatasetProperties().fieldDelimiter.setValue(FieldDelimiterType.OTHER);
    inputProps.getDatasetProperties().specificFieldDelimiter.setValue("|");

    // Create the runtime.
    SimpleFileIOInputRuntime runtime = new SimpleFileIOInputRuntime();
    runtime.initialize(null, inputProps);

    // Use the runtime in a direct pipeline to test.
    // TODO(rskraba): This fails for certain values of targetParallelism! To fix.
    final Pipeline p = beam.createPipeline(3);

    PCollection<IndexedRecord> readLines = p.apply(runtime);

    List<IndexedRecord> expected = new ArrayList<>();
    for (String record : inputFile.split("---")) {
        expected.add(ConvertToIndexedRecord.convertToAvro(record.split("\\Q|\\E")));
    }

    PAssert.that(readLines).containsInAnyOrder(expected);
    p.run().waitUntilFinish();
}

@Override
public Iterable<IndexedRecord> getRejectedWrites() {
    // If rejected write feedback is requested before submitting of current batch
    // then write accumulated records to provide feedback to a caller.
    // This is required due to bug in DI job which is not aware of bulk writes.
    flush();
    return rejectedWrites;
}

@Override
public PCollection<IndexedRecord> expand(PBegin begin) {
    FixedDatasetRuntime runtime = new FixedDatasetRuntime();
    runtime.initialize(null, properties.getDatasetProperties());

    // The values to include in the PCollection
    List<IndexedRecord> values = new LinkedList<>();

    if (properties.overrideValuesAction.getValue() == FixedInputProperties.OverrideValuesAction.NONE
            || properties.overrideValuesAction.getValue() == FixedInputProperties.OverrideValuesAction.APPEND) {
        if (!properties.getDatasetProperties().values.getValue().trim().isEmpty()) {
            values.addAll(runtime.getValues(Integer.MAX_VALUE));
        }
    }

    if (properties.overrideValuesAction.getValue() == FixedInputProperties.OverrideValuesAction.APPEND
            || properties.overrideValuesAction.getValue() == FixedInputProperties.OverrideValuesAction.REPLACE) {
        properties.getDatasetProperties().values.setValue(properties.overrideValues.getValue());
        if (!properties.getDatasetProperties().values.getValue().trim().isEmpty()) {
            values.addAll(runtime.getValues(Integer.MAX_VALUE));
        }
    }

    if (values.size() != 0) {
        PCollection<IndexedRecord> out = (PCollection<IndexedRecord>) begin
                .apply(Create.of(values).withCoder((AvroCoder) AvroCoder.of(runtime.getSchema())));
        if (properties.repeat.getValue() > 1) {
            PCollectionList<IndexedRecord> merged = PCollectionList.of(out);
            for (int i = 2; i < properties.repeat.getValue(); i++)
                merged = merged.and(out);
            out = merged.apply(Flatten.<IndexedRecord> pCollections());
        }
        return out;
    } else {
        return begin.apply(RowGeneratorIO.read().withSchema(runtime.getSchema()) //
                .withSeed(0L) //
                .withPartitions(1) //
                .withRows(properties.repeat.getValue()));
    }
}

/**
 * Evaluate one specific criteria against the given indexed record.
 * 
 * @param criteria the criteria to evaluate.
 * @param record the value to evaluate against the criteria.
 * @return whether the record should be selected for this specific criteria.
 */
private boolean evaluateCriteria(FilterRowCriteriaProperties criteria, IndexedRecord record) {
    // This is the logical operation applied to multiple values applied inside ONE specific filter criteria.
    // When using a complex av expression, one accessor can read multiple values.
    // (i.e. ALL means that all values must evaluate to true.)
    LogicalOpType fieldOp = LogicalOpType.ALL;

    // Starting point for aggregating the logical operations.
    boolean aggregate = fieldOp.createAggregate();

    String accessor = criteria.columnName.getStringValue();
    if (StringUtils.isEmpty(accessor)) {
        return false;
    }

    List<Object> values = getInputFields(record, accessor);

    if (ConditionsRowConstant.Function.COUNT.equals(criteria.function.getStringValue())) {
        values = Arrays.asList((Object) values.size());
    } else if (values.size() == 0) {
        // If the function is not COUNT and no values are returned, then consider the criteria not matched.
        return false;
    }

    // Apply all of the criteria.
    for (Object value : values) {
        aggregate = fieldOp.combineAggregate(aggregate, checkCondition(value, criteria));
        if (fieldOp.canShortCircuit(aggregate))
            break;
    }

    return aggregate;
}

/**
 * Handles response according status code
 * See Jira REST documentation for details
 * 
 * @param response Jira response, which contains status code and body
 * @param resourceToCreate JSON of resource to be created
 * @param record current {@link IndexedRecord}
 * @throws IOException in case of status code is not 201 CREATED
 */
private void handleResponse(JiraResponse response, String resourceToCreate, IndexedRecord record) throws IOException {
    int statusCode = response.getStatusCode();
    String responseError = response.getBody();
    switch (statusCode) {
    case SC_CREATED: {
        LOG.debug("Successfully created {}", resourceToCreate);
        result.successCount++;
        break;
    }
    case SC_BAD_REQUEST: {
        LOG.debug("Input is invalid {}", resourceToCreate);
        throw createRejectException("error.invalidRecordCreate", resourceToCreate, responseError);
    }
    case SC_UNAUTHORIZED: {
        LOG.debug("User is not authenticated. {} wasn't inserted", resourceToCreate);
        throw createRejectException("error.unauthorizedCreate", resourceToCreate, responseError);
    }
    case SC_FORBIDDEN: {
        LOG.debug("User does not have permission to create {}", resourceToCreate);
        throw createRejectException("error.forbiddenCreate", resourceToCreate, responseError);
    }
    default: {
        LOG.debug("Unexpected status code");
    }
    }
}

private SaveResult[] insert(IndexedRecord input) throws IOException {
    insertItems.add(input);
    if (insertItems.size() >= commitLevel) {
        return doInsert();
    }
    return null;
}

protected List<IndexedRecord> writeRows(SnowflakeConnectionTableProperties props, List<IndexedRecord> outputRows)
        throws Exception {
    TSnowflakeOutputProperties outputProps = getRightProperties(props);
    outputProps.outputAction.setValue(TSnowflakeOutputProperties.OutputAction.INSERT);
    writeRows(makeWriter(outputProps), outputRows);
    return readAndCheckRows(props, outputRows.size());
}

private static String generateInputJSON(Schema inputSchema, IndexedRecord inputIndexedRecord) throws IOException {
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    DatumWriter<IndexedRecord> writer = new GenericDatumWriter<IndexedRecord>(inputSchema);
    JsonEncoder encoder = EncoderFactory.get().jsonEncoder(inputSchema, baos, false);
    writer.write(inputIndexedRecord, encoder);
    encoder.flush();
    baos.flush();
    return new String(baos.toByteArray(), StandardCharsets.UTF_8);
}