Repository: hashgraph/hedera-services
Issue: Export account balances every exportPeriodSecs since the epoch (in consensus time)

**Summary of the defect**
The account balances file should be exported each time the signed state timestamp enters a new "period", where the timeline is broken into periods of length `balances.exportPeriodSecs` beginning with the epoch.

Also:

- [ ] Remove the timestamp for client saved .pb file

Changed files:
test-clients/src/main/java/com/hedera/services/bdd/spec/HapiApiSpec.java
hedera-node/src/test/java/com/hedera/services/state/exports/SignedStateBalancesExporterTest.java
hedera-node/src/main/java/com/hedera/services/state/exports/SignedStateBalancesExporter.java

diff --git a/hedera-node/src/main/java/com/hedera/services/state/exports/SignedStateBalancesExporter.java b/hedera-node/src/main/java/com/hedera/services/state/exports/SignedStateBalancesExporter.java
index e50a3e7329..b669f61201 100644
--- a/hedera-node/src/main/java/com/hedera/services/state/exports/SignedStateBalancesExporter.java
+++ b/hedera-node/src/main/java/com/hedera/services/state/exports/SignedStateBalancesExporter.java
@@ -94,7 +94,8 @@ public class SignedStateBalancesExporter implements BalancesExporter {
 	private String lastUsedExportDir = UNKNOWN_EXPORT_DIR;
 	private BalancesSummary summary;
 
-	Instant periodEnd = NEVER;
+	Instant periodBegin = NEVER;
+	private final int exportPeriod;
 
 	static final Comparator<SingleAccountBalances> SINGLE_ACCOUNT_BALANCES_COMPARATOR =
 			Comparator.comparing(SingleAccountBalances::getAccountID, ACCOUNT_ID_COMPARATOR);
@@ -107,21 +108,21 @@ public class SignedStateBalancesExporter implements BalancesExporter {
 		this.signer = signer;
 		this.expectedFloat = properties.getLongProperty("ledger.totalTinyBarFloat");
 		this.dynamicProperties = dynamicProperties;
+		exportPeriod = dynamicProperties.balancesExportPeriodSecs();
 	}
 
 	@Override
 	public boolean isTimeToExport(Instant now) {
-		if (periodEnd == NEVER) {
-			periodEnd = now.plusSeconds(dynamicProperties.balancesExportPeriodSecs());
-		} else {
-			if (now.isAfter(periodEnd)) {
-				periodEnd = now.plusSeconds(dynamicProperties.balancesExportPeriodSecs());
-				return true;
-			}
+		if (periodBegin != NEVER
+				&& now.getEpochSecond() / exportPeriod != periodBegin.getEpochSecond() / exportPeriod) {
+			periodBegin = now;
+			return true;
 		}
+		periodBegin = now;
 		return false;
 	}
 
+
 	@Override
 	public void exportBalancesFrom(ServicesState signedState, Instant when) {
 		if (!ensureExportDir(signedState.getNodeAccountId())) {
@@ -136,12 +137,14 @@ public class SignedStateBalancesExporter implements BalancesExporter {
 					when, summary.getTotalFloat(), expectedFloat)); }
 		log.info("Took {}ms to summarize signed state balances", watch.getTime(TimeUnit.MILLISECONDS));
 
-		if (exportCsv) {
-			toCsvFile(when);
-		}
+		// .pb account balances file is our focus, process it first to let its timestamp to stay close to
+		// epoch export period boundary
 		if (exportProto) {
 			toProtoFile(when);
 		}
+		if (exportCsv) {
+			toCsvFile(when);
+		}
 	}
 
 	private void toCsvFile(Instant exportTimeStamp) {
diff --git a/hedera-node/src/test/java/com/hedera/services/state/exports/SignedStateBalancesExporterTest.java b/hedera-node/src/test/java/com/hedera/services/state/exports/SignedStateBalancesExporterTest.java
index 5f9aecfe6f..5c8d4b5793 100644
--- a/hedera-node/src/test/java/com/hedera/services/state/exports/SignedStateBalancesExporterTest.java
+++ b/hedera-node/src/test/java/com/hedera/services/state/exports/SignedStateBalancesExporterTest.java
@@ -20,7 +20,6 @@ package com.hedera.services.state.exports;
  * ‍
  */
 
-import com.google.protobuf.InvalidProtocolBufferException;
 import com.hedera.services.ServicesState;
 import com.hedera.services.config.MockGlobalDynamicProps;
 import com.hedera.services.context.properties.GlobalDynamicProperties;
@@ -36,7 +35,6 @@ import com.hederahashgraph.api.proto.java.AccountID;
 import com.hedera.services.stream.proto.AllAccountBalances;
 import com.hedera.services.stream.proto.SingleAccountBalances;
 import com.hedera.services.stream.proto.TokenUnitBalance;
-import com.hederahashgraph.api.proto.java.TokenBalances;
 import com.hederahashgraph.api.proto.java.TokenID;
 
 import com.swirlds.common.Address;
@@ -56,8 +54,6 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.Base64;
-import java.util.Base64.Decoder;
 import java.util.Comparator;
 import java.util.List;
 import java.util.function.UnaryOperator;
@@ -626,17 +622,29 @@ class SignedStateBalancesExporterTest {
 	public void initsAsExpected() {
 		// expect:
 		assertEquals(ledgerFloat, subject.expectedFloat);
-		assertEquals(SignedStateBalancesExporter.NEVER, subject.periodEnd);
 	}
 
 	@Test
 	public void exportsWhenPeriodSecsHaveElapsed() {
-		assertFalse(subject.isTimeToExport(now));
-		assertEquals(now.plusSeconds(dynamicProperties.balancesExportPeriodSecs()), subject.periodEnd);
+		subject = new SignedStateBalancesExporter(properties, signer, dynamicProperties);
+		Instant startTime = Instant.parse("2021-03-11T10:59:59.0Z");
+		assertFalse(subject.isTimeToExport(startTime));
+		assertEquals(startTime, subject.periodBegin);
+		assertTrue(subject.isTimeToExport(startTime.plusSeconds(1)));
+		assertEquals(startTime.plusSeconds(1), subject.periodBegin);
+
+		shortlyAfter = startTime.plusSeconds(dynamicProperties.balancesExportPeriodSecs() / 2);
 		assertFalse(subject.isTimeToExport(shortlyAfter));
-		assertEquals(now.plusSeconds(dynamicProperties.balancesExportPeriodSecs()), subject.periodEnd);
+		assertEquals(shortlyAfter, subject.periodBegin);
+
+		Instant nextPeriod  = startTime.plusSeconds(dynamicProperties.balancesExportPeriodSecs() + 1);
+
+		assertTrue(subject.isTimeToExport(nextPeriod));
+		assertEquals(nextPeriod, subject.periodBegin );
+
+		anEternityLater = startTime.plusSeconds(dynamicProperties.balancesExportPeriodSecs() * 2 + 1);
 		assertTrue(subject.isTimeToExport(anEternityLater));
-		assertEquals(anEternityLater.plusSeconds(dynamicProperties.balancesExportPeriodSecs()), subject.periodEnd);
+		assertEquals(anEternityLater, subject.periodBegin);
 	}
 
 	@AfterAll
diff --git a/test-clients/src/main/java/com/hedera/services/bdd/spec/HapiApiSpec.java b/test-clients/src/main/java/com/hedera/services/bdd/spec/HapiApiSpec.java
index 0c459c9380..b49cddbdd8 100644
--- a/test-clients/src/main/java/com/hedera/services/bdd/spec/HapiApiSpec.java
+++ b/test-clients/src/main/java/com/hedera/services/bdd/spec/HapiApiSpec.java
@@ -146,13 +146,8 @@ public class HapiApiSpec implements Runnable {
 
 
 	public void exportAccountBalances(Supplier<String> dir) {
-		Instant now = Instant.now();
-		Timestamp.Builder timeStamp = Timestamp.newBuilder();
-		timeStamp.setSeconds(now.getEpochSecond())
-				.setNanos(now.getNano());
-
 		AllAccountBalances.Builder allAccountBalancesBuilder = AllAccountBalances.newBuilder()
-				.addAllAllAccounts(accountBalances).setConsensusTimestamp(timeStamp);
+				.addAllAllAccounts(accountBalances);
 
 		try (FileOutputStream fout = new FileOutputStream(dir.get())) {
 			allAccountBalancesBuilder.build().writeTo(fout);