Google Cloud BigQuery Failed Inserts, No Errors

[sean-walsh]

An akka.io customer requested help with this issue and provided sample code and config. After some review we determined the usage looked good. The alpakka connection to BigQuery works properly with good config for BigQuery batching, just not streaming inserts.

Versions used

Akka version: 2.9.3
Alpakka BigQuery version: 9.0.1
Akka Http Version: 10.6.3

Expected Behavior

Any failure to connect or insert should fail the stream.

Actual Behavior

Failure is swallowed, no write is performed.

[sean-walsh]

I was able to reproduce this. I have my service account and key set up and configured correctly because querying does work.

First I tried in the format of the documentation. There are no errors or responses and no writes to BigQuery.

public class EntityToSave {
    final private String entityId;
    final private String value;

    public EntityToSave(String entityId, String value) {
        this.entityId = entityId;
        this.value = value;
    }

    @JsonCreator
    public EntityToSave(@JsonProperty("f") JsonNode fields) {
        entityId = fields.get(0).get("v").textValue();
        value = fields.get(1).get("v").textValue();
    }

    public String getEntityId() {
        return entityId;
    }

    public String getValue() {
        return value;
    }
}

Marshaller<TableDataInsertAllRequest<EntityToSave>, RequestEntity> tableDataInsertAllMarshaller =
            BigQueryMarshallers.tableDataInsertAllRequestMarshaller();

    var inserts = List.of(new EntityToSave("entity-id-1", "value-1"));

    Sink<List<EntityToSave>, NotUsed> insertionSink =
            BigQuery.insertAll(
                   datasetId,
                   tableId,
                    InsertAllRetryPolicy.never(),
                    Optional.empty(),
                    tableDataInsertAllMarshaller);

    Source.single(inserts)
            .runWith(insertionSink, system);

I then tried to async insert and did get an error: "java.util.concurrent.ExecutionException: akka.stream.alpakka.google.ResumableUpload$InvalidResponseException: No Location header"

var inserts = List.of(new EntityToSave("entity-id-1", "value-1"));

Flow<EntityToSave, Job, NotUsed> loadFlow =
            BigQuery.insertAllAsync(datasetId, tableId, Jackson.marshaller());

    Source.from(inserts)
            .via(loadFlow)
            .runWith(Sink.seq(), system);