[Bug]: Unable to use applyRowMutations() in bigquery IO apache beam java

### What happened?

Create two tables in bigquery with primary key on table1 and table2 then create simple pipeline to read from table1 and upsert into table 2 with same schema

The primary key is will be a string with random unique value

```
PCollection rows1 = p.apply(BigQueryIO.readTableRows().from(" project.dataset.table1"));

PCollection rowmutate = rows1.apply(MapElements
.into(new TypeDescriptor() {
})
.via(tableRow -> RowMutation.of(tableRow,
RowMutationInformation.of(RowMutationInformation.MutationType.UPSERT, new Random().nextLong(1,1000000)))));

rowmutate.setCoder(RowMutation.RowMutationCoder.of())

.apply(BigQueryIO.applyRowMutations()
.to(" project.dataset.table2")
.withSchema(createSchema())
.withMethod(BigQueryIO.Write.Method.STORAGE_API_AT_LEAST_ONCE)
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER))
```
I am facing these two exceptions

`com.google.api.gax.rpc.InvalidArgumentException: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: The primary keys and clustering keys of the table are required to create an upsert stream. Entity: projects/myproject/datasets/mydataset/tables/mytable/streams/_default`

`Caused by: java.lang.RuntimeException: Append to stream projects/myproject/datasets/dataset/tables/table/streams/_default failed with stream doesn't exist
`

### Issue Priority

Priority: 1 (data loss / total loss of function)

### Issue Components

- [ ] Component: Python SDK
- [X] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [X] Component: IO connector
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner