Inputs SQS with Session based Windowing doesn't work

Hi,

Trying to use Beam with AWS SQS service as an input source, using Session windows.

The windows aren't executed. Code works well when the input source is Kafka:
```

// code placeholder
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm");

PipelineOptions
options = PipelineOptionsFactory.create();
AwsOptions awsOptions = options.as(AwsOptions.class);
BasicAWSCredentials
awsCreds = new BasicAWSCredentials("", "");
awsOptions.setAwsCredentialsProvider(new AWSStaticCredentialsProvider(awsCreds));
awsOptions.setAwsRegion("eu-west-1");

Pipeline
p = Pipeline.create(options);
// This example reads a public data set consisting of the complete works
of Shakespeare.
p.apply(SqsIO.read().withQueueUrl("https://sqs.eu-west-1.amazonaws.com/XXXXXXXXXXXXXXXX"))

/*Per
session windows*/
.apply(ParDo.of(new DoFn() {
@ProcessElement
public void processElement(@Element
Message element, OutputReceiver out) {
// Extract the timestamp from log entry we're currently
processing.

LOG.info("Message Body: {}", element.getBody());
out.output(element.getBody());
}
}))

//Set
windowing configuration

.apply(
"WindowIntoSessions",
Window.into(
Sessions.withGapDuration(Duration.standardSeconds(5)))
.withTimestampCombiner(TimestampCombiner.END_OF_WINDOW)
//.triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow()))
//
Late data is dropped
.accumulatingFiredPanes()
.withAllowedLateness(Duration.ZERO))

//Extract and
count: Extracts a the object to an KV store of
.apply(
MapElements.into( TypeDescriptors.kvs(TypeDescriptors.strings(),
TypeDescriptors.integers()))
.via(
(String testO) -> KV.of(testO, new Integer(1))
)
)

.apply("CountElements",
Sum.integersPerKey())

.apply("Log", ParDo.of(new FilterTextFn()))

.apply(
MapElements.into(TypeDescriptors.strings())
.via(
(KV wordCount) ->
wordCount.getKey() + ": " + wordCount.getValue()))
;

p.run().waitUntilFinish();
}

```

Imported from Jira [BEAM-7498](https://issues.apache.org/jira/browse/BEAM-7498). Original Jira may contain additional context.
Reported by: esteveavi.

Start with the SqsIO input path and its interaction with Beam session windows, using the supplied Kafka comparison as the behavioral reference. Check how message timestamps and watermarks are handled before Window.into(Sessions...), then reproduce the example and verify that session windows execute and produce the expected counts.

Written by the indexing model from the issue text.