text:
Welton (; or Welton by Lincoln) is a large village and civil parish in the West Lindsey district of Lincolnshire, England. The population of the civil parish was recorded as 4,327 in the 2011 census. It is geographically situated  north from Lincoln city centre. The name Welton by Lincoln is also used to distinguish from other similarly named villages in Lincolnshire: Welton le Wold and Welton le Marsh.

The village centre has been long appreciated for its picturesque and quintessentially English qualities, boasting multiple Grade II or higher listed buildings, alongside the eponymous Welton Beck, whose venerable spring really puts the 'Wel' in Welton. It has also been known as a wildlife haven due to an abundance of green space, and as a charitable place, especially through the church and schools. Since November 1969, the village has been a designated conservation area.

Toponymy
The name means, roughly, "town with a stream", with the suffix 'ton' being from the Saxon term 'tūn' for an enclosure, and 'wel' coming from the Anglo-Saxon 'wella' a place of springing or bubbling waters, possibly referencing the nearby Old Man's Head spring, the source of the Welton Beck. For this reason, it is unlikely to refer to the pump and formerly well on the southern green.

Historically, the village has been known by many diverse names and spelling variations. Most famously in 1086, the Domesday Book registered the area as 'Welletone'. Later names include 'Welletonam' as early as 1070, 'Welton Davy Bekhall' in 1291, 'Welleton cum Kirketon' in 1292, 'Wolteme Brynthall' in 1349, and 'Welton Askeby Payneshill' in 1428. Many refer to the prebends, as they were sometimes seen as similarly significant to the town itself. In relation to the neighbouring hamlet of Dunholme, the area was also called 'Welton iuxta Downeham' and 'Welton nigh Dunham' in 1583 and 1661 respectively, amongst other names.

The town has also been known as Welton by Lincoln or Welton-by-Lincoln, amongst other vaiations, to clearly differentiate it from the multitude of other settlements with a variation of the name Welton, including four within just : Welton with Melton, Welton le Wold Welton le Marsh, and Little Welton, near Louth.

History
The village does have an ancient well, later completed with a Victorian pump. Though the more famous pump is located on a small grassy triangle between Sudbeck Lane and Lincoln Road, there are other old wells with their own functioning pumps elsewhere. Before the Anglo-Saxon inhabitation, there is evidence of both Roman and Celtic settlement from about 7,500 years ago, with coins and fragments of pottery serving as evidence of a Roman farmstead. An evaluation undertaken immediately south of the parish church and near to a known Saxon cemetery, revealed Roman remains and artefacts, and recovered fragments of early-mid Saxon and medieval pottery in 2007. Another evaluation in 2007 at Heathclinic House recorded a late Saxon ditch and a series of medieval pits and ditches. Artefacts from the 12th century and later were noted to be fairly abundant, indicating probable domestic occupation of the site at that time.

Roman building debris, pottery scatter, stamped pottery, stamped tiles, and Roman coins were found on numerous occasions. Air photographs also indicated Roman site around the former hamlet. An Anglo-Saxon inhumation cemetery, a Bronze Age tanged and barbed arrowhead found, and a Neolithic stone axe were all found in Ryland too.

By 1086, the Domesday Book recorded as many as 52 households, and named the village "Welletone". It was at this time that the medieval fishpond complex beside the Welton Beck was in use.

Welton School was built by subscription in 1826, and by 1881 it had about 130 students. In 1889, it was rebuilt to accommodate 167 students. The site of the school has since become residential, but can be remembered by the symbolically named Old School Yard Passage directly adjacent to the building. In 1881, the parish was entitled to send eleven boys to Christ's Hospital (or Blue Coat) School in Lincoln.

The vicarage was recorded as discharged in 1848, meaning that as it valued in the king's books at 7 pounds, 6 shillings, and 8 pence, it fell below £10 and was therefore exempt from having to pay the First Fruits and Tenths according to law created by Henry VIII, and confirmed by Elizabeth I.

In 1863, a number of apparently Saxon graves were discovered on the site of an old chapel cemetery. Roman remains have also said to have been found in the village as well.

Bastardy cases would be heard in the Lincoln petty session hearings on the first and third Fridays of each month.

There is significant recorded personal charity for the poor through the local parish church, St Mary's, as well a history of other resources for the poor. In 1664, Thomas Codd left 20 shillings per year for the poor, and four year later in 1668, a Mrs Leary left 10 shillings per year. However, by 1881, all knowledge as to what land generated Leary's income was lost and the grant left undone. In 1716, Elizabeth Croft left 20 shillings a year to the poor. The Common Lands were enclosed in the parish around 1771. In 1824, John Camm left the interest from £500 for the poor, which was combined with about £11, 10 shillings a year recorded as chiefly the gift of Earl Brownlow by 1848, to supply about £20 to the poor per annum. Renowned philanthropist Sarah Greville, Countess of Warwick left £10 per annum to the vicar of Welton for delivering a lecture in the church every Sunday evening. The 1834 Poor Law Amendment Act made the parish part of the Lincoln Poor Law Union, and in 1881, it still held the right to send one poor man to Market Rasen Hospital.

During the Cold War, a Royal Observer Corps underground monitoring post for nuclear reporting was operated in Ryland. A similar post can be found at the site of the former RAF Dunholme Lodge to the south-west.

William Farr School was opened inside the parish as a secondary modern school in 1952, on a site purchased for £600 from RAF Dunholme Lodge in 1946 by Rev William Farr, then vicar of Welton. The school was named after him when he died in 1955. Five years later in 1960, the old former wartime buildings were replaced, with the new site straddling the southern parish border with Dunholme. The school acquired comprehensive status in 1974, became grant-maintained in 1992, and as part of their trip to Welton in 1996, Queen Elizabeth II and Prince Philip visited to open the new Humanities building.

In 2000 William Farr signed up for the latest education initiative and attained Technology College status. In 2001 they achieved the distinction of having the best comprehensive school A-level results in England, and in 2006 it received an outstanding award in every category in an Ofsted inspection, one of the best in the country. It is now an academy and is also an associate school of the University of Lincoln.

Welton also has a very successful not-for-profit magazine, Welton News, which has published monthly since June 1999, nearing 300 issues as of 2022. More than 2,000 copies of each issue are distributed free-of-charge to every household in Welton by a group of volunteers. The magazine was edited by Hugh Gilfedder for over 15 years, until his retirement in 2019, when he was succeeded by Dorothy Russell. Gilfedder was also responsible for the donating two new trees to be planted in the village the same year.

The village floral displays were planted in a purple, cream and green theme in 2018 to reflect the colours of the Suffragette movement, commemorating the 100th anniversary of when women first won the vote.

1987 Red Arrows crash

On 16 November 1987 at 12:45 pm, two of RAF Scampton's Red Arrows, Hawks XX259 and XX241, called Reds 1 and 2, piloted by Flt Lt MJ "Spike" Newbury and Sqn Ldr Tim Miller respectively, experienced a mid-air wing tip collision during a winter display practice. Red 2 collided with Red 1 after the air brake failed to operate. Red 1 slammed into houses on Monce Close and Red 2 landed in a field, just metres away from a row of houses. Both pilots ejected, but Sqdn Ldr Miller suffered injuries to his back, and Flt Lt Newbury suffered a broken leg - both had to be taken to Lincoln County Hospital for treatment. Amazingly, nobody was killed in the incident, as the occupants of the house on Monce Close at the epicentre of the crash were all out at the time of the accident. Gillian Schooley had gone to Lincoln while her young daughter was at play school. One neighbour, James Bray was in the bath when the plane crashed next door, while another, Charles Gillon, said he dived for cover after seeing the planes collide, roll over, and burst into flames from his bedroom window.

Both aircraft had to be written off after the accident due to their near total destruction. Red 2 had been on loan to the Red Arrows from No. 4 Flying Training School RAF.

In the immediate aftermath of the crash, there were calls for the Red Arrows to be barred from practising over populated areas, as the team had only been stationed at RAF Scampton for three years. Welton Parish Council chairman Malcolm Parish supported the relocation of the team. Some suggested the team practise over the sea, but were quickly dismissed. Many of Lincolnshire's members of parliament, including Gainsborough MP Sir Edward Leigh, insisted all families affected by the crash should be fully compensated.

Due to being the first in a spree of accidents involving the Red Arrows, with one three months later leading to the death of a pilot and the banning of any future aerobatic displays by the group, some suggested that the Red Arrows were jinxed.

Prebends of Welton
When Lincoln Cathedral was first built, Welton's owner, William the Conqueror, gave the parish to the bishop to endow six prebends which provided income to support six canons attached to the cathedral. These were subsequently confirmed by William II and Henry I. The prebends were:

 Beckhall
 Brinkall
 Gorehall
 Painshall
 Rivehall
 Westhall

The "H" preceding "all" is silent and not to be pronounced.

Some of the roads in the village have been named after them: Brinkhall Way, Beckhall, Rivehall Avenue, Westhall Road, Painshall Close, Gorehall Drive, and even Prebend Lane. However, there is not necessarily any geographical connection between the original prebends and the streets named after them.

Toponymy
These would generally have been named after manors, hence the suffix 'hall'. Despite Gorehall's particularly gruesome sounding name, it in fact simply originates from gāra, "a triangular plot of ground". Rivehall is described as 'manor-house associated with the reeve', a form of local magistrate and real estate manager. Beckhall is a compound of 'bekkr' and 'hall', meaning the manor by the beck. Brinkhall is formed from brink, as in 'the edge of a bank' and hall. Painshall is believed to come from an individual's name, which may have been Pain, or from historical names for the prebend, it could be Paganus, Pene, Pane, Pan, Panus, Payne, Pounce, or Paunch. Westhall is simply named for being located in the west, and has never had any other known names or spelling variations.

Attractions
The Church of St Mary is by far the most well known landmark in the village, with its long, interesting history and numerous listed structures including the celebrated war memorial. There have also been chapels in Welton for the Wesleyan Methodists, the Free Church, the Primitive Methodists and the Reform Methodists, but now only the Wesleyan chapel is still in use, known as the Welton & Dunholme Methodist Church.

St Mary's Church

The village church, the Prebendal Church of St Mary, was originally Norman, but only a few fragments of this survive in the tower walls; the present building is in the Early English style from around 1250. Apart from the pillars and arches within the church, most of the church is not the original Norman one. After it burnt down in 1442, the tower was rebuilt in 1768 and the body of the church in 1823 and 1824. The large cemetery to the south and east has likely been in use since as long as the church, though most graves from earlier than the 1700s have been lost to time, with tragically few surviving from even as far back as that. The choir vestry was added onto the north side in 1921 in memory of Bishop Edward King. Above and behind the baptismal font, on the west wall of the tower, hang the hatchments, the Royal Coat of Arms of King George III, dated 1838, which interestingly, in the year after William IV's death and Victoria's accession, do not carry the name of the monarch. Below the arms is an oil painting of the Holy family by Florentine artist Carlo Falcini Depinse, dated 1847, given to the church in 1925. Until its removal in 1876, there was a gallery up against the west wall, which would have been used by the orchestra to provide music for the services until 1851. The box pews were removed in 1888 and 1890, at the same time as the chancel was refitted, replaced with pine benches which now seat 250. The total cost of all these improvements came to £613.

The six church bells were cast by Henry Harrison, nephew of John Harrison the carpenter, clockmaker, and inventor of the marine chronometer. Henry was born in 1732 and cast the Welton bells in 1770. He was also commissioned to hang the bells at York Minster in 1733.

The church was struck by lightning in the autumn of 1847, and one person killed, and many others injured. In year, the south-western pinnacle atop the bell tower fell into the churchyard during a storm, though it was replaced shortly after. The building became Grade II* listed on 30 November 1966.

A copy of the Act of Parliament for Welton, and the original Enclosure Award, are in the church chest. The Anglican parish register dates from 1562 for baptisms, 1568 for marriages and 1575 for burials.

The Reverened William Farr was buried in the church yard after his death in the 1955, with his wooden gravestone making a small landmark at the church. The wooden marker was restored by year 12 students at William Farr School in 2021.

The tall ashlar cross near the war memorial in St Mary's churchyard, was erected in 1910 in memory of Dr Richard Smith, founder of Lincoln Bluecoats School, and became Grade II listed on 21 June 1985. Smith was born in Welton in about 1533, and later died there in 1602. He was also notable for being the physician to Lord Burleigh at the court of Elizabeth I. Smith also has a brass plaque and a stained glass window in the church dedicated to his memory, which was designed and made by Burlison & Grylls in 1917.

The other large stained glass window is a memorial window to the RAF, also designed and made by Burlison & Grylls in London, in 1919 and unveiled by the Bishop of Lincoln on 10 May 1921.

The other large stained glass window is a memorial window to the men and women of the Royal Air Force, Royal Flying Corps, and the Royal Naval Air Service, also designed and made by Burlison & Grylls in London, in 1919 and unveiled by the Bishop of Lincoln on 10 May 1921. It is said to be one of the oldest stained windows dedicated to the British Armed Forces.

A third listed structure in the churchyard is a pair of ashlar headstones to Robert Camm, who died 1781, and his wife Elizabeth, who died 1788. The two became jointly Grade II listed on 21 June 1985.

On 17 September 2014, a Black Mulberry tree was planted in the churchyard by Charles Hood, Deputy Lieutenant on behalf of HM The Lord-Lieutenant of Lincolnshire in commemoration of those who served in the First World War.

The Welton PCC is now also creating a remembrance garden in 2015 in the south-east corner of the churchyard after being awarded nearly £1,812 from the Lincolnshire Co-op's Community Champions award scheme.

Nowadays, the church is still very active with many charitable events such as the Welton Larder, with free food for any in need, even providing free delivery, and spreading awareness for global issues such as the ongoing climate crisis. The current reverend is Rev Adam Watson, who with Rev Paul Blevins blessed the dresses sent by the Christian organisation named 'Little Dresses for Africa' which sends dresses to girls around the world. The church forms part of the benefice of Welton's St Mary's, Dunholme's St Chad's and Scothern's St Germain's churches, who come together for joint prayers under the 'Threechurchespray' scheme.

With a donation of £10, all of which goes towards the upkeep of the church, the floodlights around its base can be sponsored for an occasion.

St Mary's Church is also responsible for the running of the next-door charity shop, "The Mustard Seed".

Welton War Memorial

The war memorial in St Mary's churchyard was ceremoniously erected in 1923 to commemorate the men who died fighting in World War I. It was unveiled by Col Herbert Gordon on 26 May 1923 to a crowd of Weltoners, with prayers by the Rev T Naylor, followed by the dedication by the Venerable Archdeacon Blackie, and singing. The memorial was made by A J Tuttell and son of the Monumental Masons in Lincoln, and George William Beeton (1890-1966) was photographed by J Spencer Baldry of Lincoln as a model for masons to use in the making of Welton's war memorial. Memorials using the same reference have since been erected in various villages in Lincolnshire, including Haxey and Metheringham, as well as possibly Soham in Suffolk nearby his hometown, and was also notably used for the memorial that was damaged by the IRA in Enniskillen in 1987.

It is composed of Portland stone encased in Carrara marble with leaded lettering, and became Grade II listed on 21 June 1985, alongside other local monuments.

The inscriptions read:

A number of names are not shown on the memorial and can only be seen on the smaller brass plaque within St Mary's Church, such as those who were wounded or taken prisoner. The plaque was placed there to commemorate all from the parish who fought in the wars by Alfred Hunt, former vicar of Welton. The names only seen on the plague originally included the three men who died during World War II, although they were later added to the base of the pedestal.

The name of George Grantham, who was killed in August 1918, shortly before the memorial was built, was added onto the physical memorial in around 2005, though to a separate piece of stone at the base of the pedestal, between the encompassing flowers. Harry Cottingham's name is shown on the smaller brass memorial, but only otherwise on the war memorial in neighbouring Dunholme. Notably, out of all the names on the memorial, there is one known error, as the name of Hermann Reed is incorrectly engraved as 'Herman Read'.

The first meeting to establish a committee for the organisation of a memorial was held on 28 February 1919, followed by a second meeting on 11 March. There was a large attendance of parishioners and James Lillie was appointed chairman of the meeting. It was initially decided to have a memorial hall, but the committee did not proceed with the idea. On 24 August 1922, authorisation was granted for the erection of the present memorial – a soldier of the Lincolnshire Regiment, in full battle dress, with reversed arms. The faculty also granted post permission for the RAF memorial window inside St Mary's Church, Welton. In order to be more easily accessible, a portion of the wall around the churchyard was removed and replaced by iron gates leading up to a series of stone steps. It was said in 1923 that the memorial had cost about £200, of which £65 was still outstanding by the start of June. The scheme was carried into effect by the committee, then with Mr W Lyon as their secretary.

The former flag mast from RAF Scampton was erected beside the war memorial in 1997.

Due to a lack of any known surviving documentation referencing the party responsible for the maintenance of the war memorial, the parish council (as it is empowered to do by the War Memorials (Local Authorities' Powers) Act 1923 - s 1 & 3), formally accepted responsibility for it at a meeting on 19 May 1997. The memorial is now insured for public liability by the parish council.

As part of the 2020s Bishop's Green development to the north of the village, many new streets used names from the memorial, such as Hodson, Twell, or Reed, using the corrected spelling.

Northern green
The main green is a grassy triangular park in the village centre. It is generally known as just 'the green', but can be called the northern green to differentiate from the similar space surrounding the pump. The green is home to a mature lime tree, which was planted at the same time as an ornate cast-iron lamppost was erected in 1897 to commemorate Queen Victoria's Diamond Jubilee. A second lamp was placed on the northern side of the green to commemorate Queen Elizabeth II's Diamond Jubilee in 2012. There is also a map board and a trio of semi-mature deciduous trees, and a recently added bench encircling the Victorian tree.

Controversially, the space is also home to "less than appropriate streetscape features," such as the phone box, bus shelter, and modern street lamp.

Pump

Though there are many old pumps throughout the village, the most famous, and likely oldest pump is located on a small grassy triangle between Sudbeck Lane and Lincoln Road, often known as the second or southern green. It is believed to have been in use as a well since the Anglo-Saxons inhabited the area, only more recently with the addition of a Victorian pump.

A common mistake regarding the name of the village is that the pump, or earlier well, is responsible for the 'wel' prefix. However, since it comes from the Anglo-Saxon 'wella,' meaning 'springing or bubbling waters or stream' a more likely origin would instead be the beck itself or the source at Old Man's Head spring just west of the village. It is likely because of this misunderstanding that the pump has become a primary symbol of the village, featuring on much Welton imagery, emblems, and even the village sign.

An ornate plaque on the side of the trough in front of the pump reads "Welton-by-Lincoln Pump Refurbished 1995" with a second on its front thanking the family of Janet Spence for their support of the pump's refurbishment.

Welton & Dunholme Methodist Church
The chapel opened in 1815 and a Sunday school room was in 1898. An extension was added to the south of the original building in the 2010s. The chapel holds a Sunday worship, and their programme claims to include children's and youth clubs, men's and women's fellowships, Bible Study, a Prayer Group and many other social, community and church activities. It is now the only remaining chaple in use in Welton, with the other two from other denominations in Ryland and Welton Hill now both closed.

Geography
The town covers about , almost entirely within Welton parish, save for a few residential streets and William Farr School, straddling Dunholme parish in the south. The village centre has been long appreciated for its picturesque and quintessentially English qualities, with a multitude of Grade II or higher listed buildings, an abundance of green space, and the eponymous Welton Beck. The green spaces have also allowed the village to have a reputation as a wildlife haven. In November 1969, the most of the village was made a designated conservation area.

Welton is noted for the rare chalk formation known as the Welton Band. Ironically, the layer is not visible from Welton, but on just seven sites much further north-east.

Kelly's Directory of Lincolnshire described the physical geography of the parish in 1898, saying the soil was loamy; subsoil, oolite in the upper, clay in the lower part of the parish. The chief crops were wheat, barley and oats. The area was also described as having an abundance of good building-stone in 1848.

To the immediate south-east is the neighbouring village of Dunholme, near the A46. Former air base RAF Dunholme Lodge is situated next to both villages and was a wartime bomber base used intermittently from 1941 to 1964. It was land bought from this RAF base that William Farr School was built on. Some runway still remains, though the majority of the land is now used for farming.

Beside the church, there are a handful of boulders, some of which lie against the wall. They were carried to the village during the ice age.

Though it is now governed by the West Lindsey District Council, the parish was in the ancient Lawress Wapentake. The parish was also in the north-east sub-district of the Lincoln Registration District. The village was also formerly the centre of the Welton Rural District, a region of the Parts of Lindsey, from is establishment in the Local Government Act 1894, until it was abolished under the Local Government Act 1972, then and to this day becoming part of the West Lindsey district of Lincolnshire.

As a result of the 1834 Poor Law Amendment Act, the parish became part of the Lincoln Poor Law Union. In 1881, the parish still held the right to send one poor man to Market Rasen Hospital.

Welton Beck

Welton Beck is a small stream which originates from Old Man's Head Spring in the west of Welton parish and flows eastwards through Welton and Dunholme for approximately 6 km to a confluence with the Barlings Eau near Reasby, which goes on to join the River Witham. The beck is groundwater fed, from springs of the Lincolnshire limestone aquifer. It has clear water and aquatic plant growth characteristic of calcareous conditions. Throughout the course of the beck, the channel has been significantly modified from its natural state, having been straightened, widened, and its course altered.

Well dressing ceremonies
The beck has long been the site of traditional well dressing ceremonies. This was a tradition that involved decorating a local spring to act as a "thanksgiving to Almighty God for the blessing of a bountiful supply of pure water to Welton". The last known ceremony on a Lincolnshire beck was in 1924.

The custom was an annual event which took place on Ascension Day. Five wells in the village were dressed, including one in the churchyard, one in the grounds of the vicarage, two in West Carr, and one in spring cottage on Sudbeck Lane. The dressing of the wells took a different format to that of neighbouring counties, Derbyshire and Nottinghamshire; in Welton, each area surrounding the well was marked with an arch formed from a tree branch and decorated with lilac and laburnum. White calico cloth on which a text taken from the bible was depicted was put into each arch by the men in the village early on Ascension Day morning. The ceremony began with a service in Saint Mary's Church, followed by a parade to the decorated beck in the churchyard. Each well was then dressed in turn and a prayer said and a hymn sung. The local Sunday school children took part in the ceremony by placing wild flowers at each well.

There is believed to have been an ancient belief of healing powers of the beck water. For example, during a whooping cough epidemic in the village in the early 1900s, mothers took their prams containing the infants and stood them in the beck, believing that the germs would be carried away with the flow of the fresh water.

Government

Welton civil parish

The civil parish of Welton covers about , hemmed in by Hackthorn parish to the north, and Dunholme parish to the south, also including a few scattered hamlets, most notably the former hamlet of Ryland in its east.

The parish stretches from the ancient Ermine Street (A15) in the west, following a north-easterly shape to the Barlings Eau at its eastern end, of which the Welton Beck is a tributary, and is an example of a strip parish. Due to the low, flat topography of Lincolnshire, land had to be drained for agriculture to be successful. The larger drainage channels, many of which are parallel to each other, became boundaries between parishes, causing the long and thin shape of the parish. Welton's northern and southern extents are less obviously defined. The south follows Horncaslte Lane from the A15, before cutting through the town of Welton and William Farr School until it meets up with a small stream which is followed most of the way to the Barlings Eau. The northern boundary starts in the west at RAF Scampton's 1950s runway extension and follows a vague path through fields until once again reaching the Barlings Eau river near Snarford.

Other than the main town of Welton, the civil parish includes several small hamlets, most notably Welton Hill and Welton Cliff, as well as the formerly separate Ryland. There is also evidence for other inhabited places by the names of Welton Heath, Welton Mill, and East Field.

Ryland
Ryland was a small hamlet to the east of Welton that was seemingly unofficially incorporated in the late 1900s as the village grew. It was also the site of Welton's Primitive Methodist Chapel, which was built in 1859 and closed 100 years later in 1959. It was built out of gault brick with a slate roof, segmental headed windows, and a pointed arch doorway, probably as a replacement for a general meeting place that had been used since 1851. It is currently used as a garage, and has had a garage door inserted into one of the gable ends.

The hamlet is also home to a variety of archaeological sites, including numerous Roman artefacts and buildings, as well as Anglo-Saxon, Bronze Age, and Neolithic sites and artefacts, all found in Ryland too.

The name Ryland is said to originate from the Old English 'ryge' and 'land', basically meaning a 'strip of land where rye was grown'. Like Welton, and most other places in England, the hamlet was known by other names and spelling variations in the past, such as 'de Riland' from at least 1160 to 1245.

The hamlet was recorded as having 97 inhabitants around 1858.

Welton Hill
In the furthest east of the parish of Welton  north-east of the main town is a small hamlet with less than twenty buildings, with attractions including the former Free Methodist chapel and Seven Districts Coffee Roasters. Seven Districts opened on 19 September 2020 in the premises of the former Farmers Arms pub on Market Rasen Road (A46), which had been empty from 2017, apart from when it was mmysteriously abandoned midway through a refurbishment to open as an Indian restaurant called the Farmers Mahal in 2018. Ben Southall and Ellis Purvis from nearby Nettleham had been roasting their coffee from an industrial unit in Reepham before relocating, and named their business for the seven different types of speciality coffee they produce, including one from the Supía, Colombia. The former Free Methodist chapel which was built alongsidethe A46 in 1866. It is a small attractive building in red and pale yellow brick with dog-tooth decoration on the front elevation.

The hamlet has also been known as 'Long Welton Hill', 'Short Welton Hill', 'Weletonhul', and 'Snarford Hill', likely as it is physically located very close to the hamlet of Snarford.

Welton Cliff
Located in the far west just by the border of the Scampton civil parish, the area is barely even a hamlet, with the only notable landmark being Cliffe Farm.

Electoral divisions

Lincolnshire county council

The electoral divisions that contained or were named for Welton varied greatly since their inception. From 1 April 1974 to 7 May 1981, Welton's first electoral divisions were Welton No. 1, Welton No. 2, and Welton No. 3, all with one councillor each and their first election being on 12 April 1973, and last in 1977.

From 7 May 1981 to 7 June 2001, Welton's electoral divisions largely joined with Waddingham and Spital's to the north along Lincoln Cliff to form the new division of Caenby with just one councillor. Caenby was used in the elections of 1981, 1985, 1989, 1993, and 1997.

However, from 7 June 2001 to the present, a new border and name was once again assigned to Welton's electoral division, this time name Welton Rural (not to be confused with Welton Rural District), which once again had just one councillor, and participated in the elections of 2001, 2005, 2009, 2013, 2017, and 2021. The division had some minor border adjustments in 2009, from changes to the constituent parishes borders, and a larger one on 13 December 2016, when the entire parishes of Lissington and Buslingthorpe reorganisation of Market Rasen Wold after the erasure of the Ancholme Cliff division.

West Lindsey district council
Welton ward was situated towards the centre of West Lindsey, south of the Waddingham & Spital and Middle Rasen wards. The ward is , comprising the parishes of Cold Hanworth, Faldingworth, Hackthorn, Spridlington, Toft Newton, and Welton. Welton ward population was recorded as 5,016 in 2001.

The ward was in use from 1 April 1974, first election 7 June 1973, to 7 May 2015, originally having just one councillor, but upping to two on 6 May 1999.

On 7 May 2015, Welton ward merged with Dunholme ward to form the Welton and Dunholme ward, with the last population estimates being at 8,222 in 2011, based on data from the two former wards.

UK Parliament constituency
Welton has voted as part of the Gainsborough constituency since the 1885 general election. It became part of Gainsborough and Horncastle when it existed from the 1983 election until 1997, after which it returned to the Gainsborough constituency.

Climate
As with the rest of the British Isles, Welton has a temperate oceanic climate (Köppen: Cfb) with cool summers and mild winters. The nearest official Met Office weather station for which online records are available is at RAF Scampton, less than  to the south of the town centre.

In a typical year, the warmest day should reach a high temperature of at least , whereas the coldest day should always reach a low temperature of .

The record high at RAF Scampton peaked at  in the afternoon of 19 July 2022 during the 2022 United Kingdom heat wave, beating not only the previous local record of  from a three-day heatwave on 26 July 2019, but also the former national record of  also from July 2019.

The absolute minimum temperature of  was recorded on 7 December 2010 at RAF Scampton, during the record-breaking winter of 2010–11 in Great Britain and Ireland. A former nearby weather station holds the record for the lowest daytime maximum temperature recorded in England in the month of December at  on 17 December 1981. In a year, 48.98 nights should register an air frost.

The length of the day varies extremely over the course of the year in Welton. The shortest day may have 7 hours and 30 minutes of daylight, and the longest as much as 10 hours more, with 17 hours of daylight. The earliest sunrise is at around 4:30 am in June, and the latest sunrise is 4 hours later at 8:30 am in December. The earliest sunset is at 3:30 pm in December, and the latest is 6 hours later at 9:30 pm in June. Daylight saving time (DST) is observed in Welton, starting in the spring, lasting about 7 months, and ending in the autumn.

Winters are generally cool with little temperature variation. Heavy snow is rare but snow usually falls at least once each winter. Spring and autumn can be pleasant.

Demography
The 2011 census recorded the population of the civil parish to be 4,327.

The 2001 census recorded a resident village population of Welton civil parish alone as 3,821. The population in the Welton ward was 5,016, which includes other smaller villages and nearby parishes from Hackthorn to Faldingworth. The population of the Welton civil parish alone was 3,821.

Amenities
There is a wide range of services, amenities, and retail outlets in the village, ranging from supermarkets to a tea shop, gift shops and takeaways including an award-winning fish and chip shop, as well as banking and financial services. There is also a campsite, Co-op store, local butchers, and a veterinary clinic.

Welton Manor Golf Centre has proved to be a major asset to the village, as a Golf Centre of Excellence with eighteen holes of challenging golf to suit all players. Its 120 acres include tree-lined fairways, lakes, and meandering streams, which eventually go on to join the Barlings Eau and link up with the Welton Beck on the way to the River Witham. The site also houses a driving range and a fishing lake with a caravan and camping site. Its club house is now the Falconer bar and restaurant, which is licensed for civil wedding ceremonies and christenings. The site originally opened in the mid-2000s as a nine-hole course, with the restaurant opening a year earlier, and is an example of successful diversification by local farmer Charles Ottewell, and his sons, Andrew and David.

Welton has two public houses: The Falconer by the golf course, and The Black Bull in the centre of town.

The Black Bull is rumoured to be a former regular of the Dam Busters, and reputedly, haunted. Though no ghost has ever been seen, the haunting has often been reported because of inexplicable sounds believed to be a phantom slowly climbing the eighteen stairs to the restaurant. The pub is run as a joint venture between equal partners Derek Wright, his wife Bridget, her son Thomas Allen, and his partner Leanne Robinson. Reportedly, special themed nights such as pie nights, steak nights and quiz nights which have proved particularly popular as well as a monthly meat raffle, and live music nights on the last Friday of the month, featuring all local bands.

The Parish Office houses a collection of historic village items and gifts received through the village's twinning on display in the lower room. The Parish Council operates a Community Award which is presented annually to a member of the parish nominated by the community for voluntary work they have carried out in Welton for the benefit of others. They also award Certificates of Achievement if there are a number of nominations, and a Group Recognition Certificate.

Marvin Liddle operates the registered charity Project Cornerstone on Ryland Road, that collects funds from foreign and old coinage donations for education in Ghana, healthcare education in Malawi and Burundi, and a crisis fund for Phuket, Thailand.

Manor Park pavilion is used twice weekly for Pilates classes, a toddler's music programme, by two bridge clubs and the bowls club, together with the football club, who regularly play at weekends. It is a popular venue for birthday parties. Welton Library was refurbished in February 2008, but moved from the premises of the Co-op beside the clinic in August 2015 due to legal complications, only reopening at the Community Hub at Manor Park in March 2016. The new location receives a regular attendance of around 40 people, despite controversially sharing its new smaller space with various other groups. Welton Library and Community Hub now only operates for half a week due to space sharing, but has still been described as a worthwhile investment in retaining the service for the village.

The barrier for vehicles entering the car park at Manor Park has a surprisingly long history of vandalism. Across 2019, there were reports of vandalism, accidental damage and mechanical failures, where the newly installed number plate recognition camera was said to have been instrumental in identifying the culprits of anti-social behaviour and vehicles being used irresponsibly on the park. In 2021, three vehicles were locked in at Manor Park after the barrier closed at 10:30 pm, despite a "prominent notice warning of the closure time." One owner attempted to raise the barrier, one rang the parish office, and another rammed the barrier with their vehicle.

The village playing field serves as the secondary recreational grounds in Welton, despite its more central location. There is a children's playground and the Sports and Social Club, where members play darts and pool as well as outdoor activities such as football and cricket.

The Rednil Farm Equestrian Centre, established 1980, is a 60-acre riding centre with its own cross-country course as well as indoor and outdoor manège, a jumping arena, school and social room, offices, tack rooms and stables. It teaches riding in different disciplines and at different levels and also trains and corrects horses and ponies.

Welton has two schools, the Ofsted outstanding William Farr secondary school on Lincoln Road, and St Mary's C of E Primary Academy on School Drive. The students at the primary school designed a painted model of the Lincoln Imp that was placed at the Lincoln Hotel for the Lincoln Imp Trail 2021. In conjunction with the school, the related Friends of St Mary's school group of parents and staff also organises events frequently, such as 22 to 31 October Scarecrow Trail around the village, and yearly Easter bingo, though many plans were disrupted by the COVID-19 pandemic. There are also three preschools.

The village has two youth football teams, the U12 Bombers and U16 Bombers, both of which play in their respective age group of the Lincolnshire Co-op Mid Lincs County Youth Football League. In conjunction with Welton Parish Council, the clubhouse at Manor Park maintains a record of top scorers.

The Scout Hut provides activities for Beavers, Cubs and Scouts, Rainbows, Brownies and Guides.

International Relations

Twin towns – sister cities

Welton is twinned with:

 Moncé-en-Belin, France (since 1974)

Welton and Moncé-en-Belin signed the Official Twinning Charter on 28 September 1974 in Welton. The twinning committee oversees all activity related to twinning with the council in Moncé-en-Belin, chaired by Cllr Marlene Chapman, vice-chair of Welton's council, who usually gives the welcome speech on the Moncéen arrival in Welton. The two councils meet and celebrate anniversaries every 10 years. The most recent anniversary was marked in summber 2014 when the Parish Council hosted a reception for the Frecnh visitors for the 40th anniversary. Visits between members of the Family Twinning Association take place each year over the August bank holiday weekend with the people for Moncé-en-Belin visiting Welton every even year, and vice versa every odd. There is a French post box on the wall of the Parish Council office, and a street named Monce Close in Welton, as well as an Avenue de Welton in Moncé-en-Belin.

Footnotes

References

External links

Welton Village Website
Welton Chalk formations
Primary school
Welton by Lincoln Parish council website
Lincoln Imp Trail 2021

Villages in Lincolnshire
Civil parishes in Lincolnshire
West Lindsey District


If you ever stopped to listen to all the chatter in your head, you would hear many thoughts of self-loathing and not many thoughts oozing with Self-esteem. Why is loving yourself so hard?
Because not loving ourselves has been our collective destiny until now. For thousands of years, women have been diminished and persecuted – leading to lifetimes of suffering at the hands of others and ourselves.
Yet, in this lifetime, spiritual women have a deep knowing that the Divine co-creates our lives with us and it will end well for us. Quite simply, suffering and persecution are no longer required. Join Robin Clare to gain a deeper understanding of how our collective past has influenced our present and what is possible for our future.


Your Akron Fleming's is located on the corner of Route 18 (Medina Road) and Flight Memorial Parkway, just east of Route 77N in the Montrose Shopping area next to Montrose Ford. Whether you're celebrating, dining with business clients or want a night out with friends, we have created the perfect ambiance with a menu filled with your favorites and a local chef's table that changes with the season, or whenever we feel inspired by our ingredients. Our bar is always open and our wine manager has curated a wide selection of locally selected wines, hand-crafted cocktails and spirits. Need a more private space? Our Private Dining Director is at your service.
Welcome to Your Akron Fleming's.


text:
Home » News & Speeches » Record number of projects submitted to the 2015 World Chambers Competition
Record number of projects submitted to the 2015 World Chambers Competition
The International Chamber of Commerce (ICC) World Chambers Federation (WCF) has received a record number of innovation and best practice projects for the World Chambers Competition, a prominent feature of the ICC WCF 9th World Chambers Congress to be held in Torino, Italy in June.
9th World Chambers Competition
Open to more than 12,000 chambers across all five continents, the Competition has received 79 applications from 40 countries for the 2015 edition, making this year's event the biggest since its launch in 2003 in Quebec City.
"The World Chambers Competition is more than just an awards programme. It is a unique forum to showcase the ingenuity behind cutting-edge business projects undertaken by small- and medium-sized enterprises that are part of chambers of commerce worldwide," said Rona Yircali, Honorary Chair of the ICC World Chambers Federation and Chair of the Competition jury.
The Competition is an unparalleled opportunity to raise the profile of these ground-breaking projects and give them significant and much deserved visibility in the world business community.
"The Competition is an unparalleled opportunity to raise the profile of these ground-breaking projects and give them significant and much deserved visibility in the world business community," Mr Yircali said.
In the coming weeks, an international jury composed of representatives from transnational, national and local chambers of commerce will shortlist four finalists for each of the following categories:
Best corporate social responsibility project
Best job creation and business development project
Best policy advocacy project
Best unconventional project
Successful chambers will be invited to present their projects to an audience of chamber delegates and business professionals during Competition finals at the 9th World Chambers Congress which will take place on 10-12 June. Winners of each category will be announced at the Congress' gala dinner on 12 June.
The 9th World Chambers Congress, taking place from 10-12 June in Torino, Italy is the largest global gathering for chambers of commerce. Held every two years in a different region of the world, the Congress is the only international forum for chamber leaders to share best-practice experiences, exchange insights, develop networks, address the latest business issues affecting their communities and learn about new areas of innovation from other chambers around the world.
9th World Chambers Congress
ICC World Chambers Federation
Follow the 9th World Chambers Congress via @WorldChambers and #9WCC.
Congress participants are invited to join the discussion on the Linkedin group.
Applications for the 2015 World Chambers Competition are now closed.
World Chambers Competition
World Chambers Congress
11th World Chambers Congress Report
World chambers report explores chamber of commerce response to COVID-19

redpajama_set_name: RedPajamaCommonCrawl


text:
package kafka.log

import java.io._
import java.util.Properties

import org.apache.kafka.common.errors.{CorruptRecordException, OffsetOutOfRangeException, RecordBatchTooLargeException, RecordTooLargeException}
import kafka.api.ApiVersion
import kafka.common.LongRef
import org.junit.Assert._
import org.scalatest.junit.JUnitSuite
import org.junit.{After, Before, Test}
import kafka.message._
import kafka.utils._
import kafka.server.KafkaConfig
import org.apache.kafka.common.utils.Utils

class LogTest extends JUnitSuite {

  val tmpDir = TestUtils.tempDir()
  val logDir = TestUtils.randomPartitionLogDir(tmpDir)
  val time = new MockTime()
  var config: KafkaConfig = null
  val logConfig = LogConfig()

  @Before
  def setUp() {
    val props = TestUtils.createBrokerConfig(0, "127.0.0.1:1", port = -1)
    config = KafkaConfig.fromProps(props)
  }

  @After
  def tearDown() {
    Utils.delete(tmpDir)
  }

  def createEmptyLogs(dir: File, offsets: Int*) {
    for(offset <- offsets) {
      Log.logFilename(dir, offset).createNewFile()
      Log.indexFilename(dir, offset).createNewFile()
    }
  }

  /**
   * Tests for time based log roll. This test appends messages then changes the time
   * using the mock clock to force the log to roll and checks the number of segments.
   */
  @Test
  def testTimeBasedLogRoll() {
    val set = TestUtils.singleMessageSet("test".getBytes)

    val logProps = new Properties()
    logProps.put(LogConfig.SegmentMsProp, (1 * 60 * 60L): java.lang.Long)

    // create a log
    val log = new Log(logDir,
                      LogConfig(logProps),
                      recoveryPoint = 0L,
                      scheduler = time.scheduler,
                      time = time)
    assertEquals("Log begins with a single empty segment.", 1, log.numberOfSegments)
    // Test the segment rolling behavior when messages do not have a timestamp.
    time.sleep(log.config.segmentMs + 1)
    log.append(set)
    assertEquals("Log doesn't roll if doing so creates an empty segment.", 1, log.numberOfSegments)

    log.append(set)
    assertEquals("Log rolls on this append since time has expired.", 2, log.numberOfSegments)

    for(numSegments <- 3 until 5) {
      time.sleep(log.config.segmentMs + 1)
      log.append(set)
      assertEquals("Changing time beyond rollMs and appending should create a new segment.", numSegments, log.numberOfSegments)
    }

    // Append a message with timestamp to a segment whose first messgae do not have a timestamp.
    val setWithTimestamp =
      TestUtils.singleMessageSet(payload = "test".getBytes, timestamp = time.milliseconds + log.config.segmentMs + 1)
    log.append(setWithTimestamp)
    assertEquals("Segment should not have been rolled out because the log rolling should be based on wall clock.", 4, log.numberOfSegments)

    // Test the segment rolling behavior when messages have timestamps.
    time.sleep(log.config.segmentMs + 1)
    log.append(setWithTimestamp)
    assertEquals("A new segment should have been rolled out", 5, log.numberOfSegments)

    // move the wall clock beyond log rolling time
    time.sleep(log.config.segmentMs + 1)
    log.append(setWithTimestamp)
    assertEquals("Log should not roll because the roll should depend on timestamp of the first message.", 5, log.numberOfSegments)

    val setWithExpiredTimestamp = TestUtils.singleMessageSet(payload = "test".getBytes, timestamp = time.milliseconds)
    log.append(setWithExpiredTimestamp)
    assertEquals("Log should roll because the timestamp in the message should make the log segment expire.", 6, log.numberOfSegments)

    val numSegments = log.numberOfSegments
    time.sleep(log.config.segmentMs + 1)
    log.append(new ByteBufferMessageSet())
    assertEquals("Appending an empty message set should not roll log even if succient time has passed.", numSegments, log.numberOfSegments)
  }

  /**
   * Test for jitter s for time based log roll. This test appends messages then changes the time
   * using the mock clock to force the log to roll and checks the number of segments.
   */
  @Test
  def testTimeBasedLogRollJitter() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val maxJitter = 20 * 60L

    val logProps = new Properties()
    logProps.put(LogConfig.SegmentMsProp, 1 * 60 * 60: java.lang.Long)
    logProps.put(LogConfig.SegmentJitterMsProp, maxJitter: java.lang.Long)
    // create a log
    val log = new Log(logDir,
      LogConfig(logProps),
      recoveryPoint = 0L,
      scheduler = time.scheduler,
      time = time)
    assertEquals("Log begins with a single empty segment.", 1, log.numberOfSegments)
    log.append(set)

    time.sleep(log.config.segmentMs - maxJitter)
    log.append(set)
    assertEquals("Log does not roll on this append because it occurs earlier than max jitter", 1, log.numberOfSegments)
    time.sleep(maxJitter - log.activeSegment.rollJitterMs + 1)
    log.append(set)
    assertEquals("Log should roll after segmentMs adjusted by random jitter", 2, log.numberOfSegments)
  }

  /**
   * Test that appending more than the maximum segment size rolls the log
   */
  @Test
  def testSizeBasedLogRoll() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val setSize = set.sizeInBytes
    val msgPerSeg = 10
    val segmentSize = msgPerSeg * (setSize - 1) // each segment will be 10 messages

    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, segmentSize: java.lang.Integer)
    // We use need to use magic value 1 here because the test is message size sensitive.
    logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString)
    // create a log
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments)

    // segments expire in size
    for (_ <- 1 to (msgPerSeg + 1)) {
      log.append(set)
    }
    assertEquals("There should be exactly 2 segments.", 2, log.numberOfSegments)
  }

  /**
   * Test that we can open and append to an empty log
   */
  @Test
  def testLoadEmptyLog() {
    createEmptyLogs(logDir, 0)
    val log = new Log(logDir, logConfig, recoveryPoint = 0L, time.scheduler, time = time)
    log.append(TestUtils.singleMessageSet("test".getBytes))
  }

  /**
   * This test case appends a bunch of messages and checks that we can read them all back using sequential offsets.
   */
  @Test
  def testAppendAndReadWithSequentialOffsets() {
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 71: java.lang.Integer)
    // We use need to use magic value 1 here because the test is message size sensitive.
    logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    val messages = (0 until 100 by 2).map(id => new Message(id.toString.getBytes)).toArray

    for(i <- 0 until messages.length)
      log.append(new ByteBufferMessageSet(NoCompressionCodec, messages = messages(i)))
    for(i <- 0 until messages.length) {
      val read = log.read(i, 100, Some(i+1)).messageSet.head
      assertEquals("Offset read should match order appended.", i, read.offset)
      assertEquals("Message should match appended.", messages(i), read.message)
    }
    assertEquals("Reading beyond the last message returns nothing.", 0, log.read(messages.length, 100, None).messageSet.size)
  }

  /**
   * This test appends a bunch of messages with non-sequential offsets and checks that we can read the correct message
   * from any offset less than the logEndOffset including offsets not appended.
   */
  @Test
  def testAppendAndReadWithNonSequentialOffsets() {
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 71: java.lang.Integer)
    val log = new Log(logDir,  LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    val messageIds = ((0 until 50) ++ (50 until 200 by 7)).toArray
    val messages = messageIds.map(id => new Message(id.toString.getBytes))

    // now test the case that we give the offsets and use non-sequential offsets
    for(i <- 0 until messages.length)
      log.append(new ByteBufferMessageSet(NoCompressionCodec, new LongRef(messageIds(i)), messages = messages(i)), assignOffsets = false)
    for(i <- 50 until messageIds.max) {
      val idx = messageIds.indexWhere(_ >= i)
      val read = log.read(i, 100, None).messageSet.head
      assertEquals("Offset read should match message id.", messageIds(idx), read.offset)
      assertEquals("Message should match appended.", messages(idx), read.message)
    }
  }

  /**
   * This test covers an odd case where we have a gap in the offsets that falls at the end of a log segment.
   * Specifically we create a log where the last message in the first segment has offset 0. If we
   * then read offset 1, we should expect this read to come from the second segment, even though the
   * first segment has the greatest lower bound on the offset.
   */
  @Test
  def testReadAtLogGap() {
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 300: java.lang.Integer)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)

    // keep appending until we have two segments with only a single message in the second segment
    while(log.numberOfSegments == 1)
      log.append(new ByteBufferMessageSet(NoCompressionCodec, messages = new Message("42".getBytes)))

    // now manually truncate off all but one message from the first segment to create a gap in the messages
    log.logSegments.head.truncateTo(1)

    assertEquals("A read should now return the last message in the log", log.logEndOffset - 1, log.read(1, 200, None).messageSet.head.offset)
  }

  @Test
  def testReadWithMinMessage() {
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 71: java.lang.Integer)
    val log = new Log(logDir,  LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    val messageIds = ((0 until 50) ++ (50 until 200 by 7)).toArray
    val messages = messageIds.map(id => new Message(id.toString.getBytes))

    // now test the case that we give the offsets and use non-sequential offsets
    for (i <- 0 until messages.length)
      log.append(new ByteBufferMessageSet(NoCompressionCodec, new LongRef(messageIds(i)), messages = messages(i)),
        assignOffsets = false)

    for (i <- 50 until messageIds.max) {
      val idx = messageIds.indexWhere(_ >= i)
      val reads = Seq(
        log.read(i, 1, minOneMessage = true),
        log.read(i, 100, minOneMessage = true),
        log.read(i, 100, Some(10000), minOneMessage = true)
      ).map(_.messageSet.head)
      reads.foreach { read =>
        assertEquals("Offset read should match message id.", messageIds(idx), read.offset)
        assertEquals("Message should match appended.", messages(idx), read.message)
      }

      assertEquals(Seq.empty, log.read(i, 1, Some(1), minOneMessage = true).messageSet.toIndexedSeq)
    }

  }

  @Test
  def testReadWithTooSmallMaxLength() {
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 71: java.lang.Integer)
    val log = new Log(logDir,  LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    val messageIds = ((0 until 50) ++ (50 until 200 by 7)).toArray
    val messages = messageIds.map(id => new Message(id.toString.getBytes))

    // now test the case that we give the offsets and use non-sequential offsets
    for (i <- 0 until messages.length)
      log.append(new ByteBufferMessageSet(NoCompressionCodec, new LongRef(messageIds(i)), messages = messages(i)),
        assignOffsets = false)

    for (i <- 50 until messageIds.max) {
      assertEquals(MessageSet.Empty, log.read(i, 0).messageSet)

      // we return an incomplete message instead of an empty one for the case below
      // we use this mechanism to tell consumers of the fetch request version 2 and below that the message size is
      // larger than the fetch size
      // in fetch request version 3, we no longer need this as we return oversized messages from the first non-empty
      // partition
      val fetchInfo = log.read(i, 1)
      assertTrue(fetchInfo.firstMessageSetIncomplete)
      assertTrue(fetchInfo.messageSet.isInstanceOf[FileMessageSet])
      assertEquals(1, fetchInfo.messageSet.sizeInBytes)
    }
  }

  /**
   * Test reading at the boundary of the log, specifically
   * - reading from the logEndOffset should give an empty message set
   * - reading from the maxOffset should give an empty message set
   * - reading beyond the log end offset should throw an OffsetOutOfRangeException
   */
  @Test
  def testReadOutOfRange() {
    createEmptyLogs(logDir, 1024)
    val logProps = new Properties()

    // set up replica log starting with offset 1024 and with one message (at offset 1024)
    logProps.put(LogConfig.SegmentBytesProp, 1024: java.lang.Integer)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    log.append(new ByteBufferMessageSet(NoCompressionCodec, messages = new Message("42".getBytes)))

    assertEquals("Reading at the log end offset should produce 0 byte read.", 0, log.read(1025, 1000).messageSet.sizeInBytes)

    try {
      log.read(0, 1000)
      fail("Reading below the log start offset should throw OffsetOutOfRangeException")
    } catch {
      case _: OffsetOutOfRangeException => // This is good.
    }

    try {
      log.read(1026, 1000)
      fail("Reading at beyond the log end offset should throw OffsetOutOfRangeException")
    } catch {
      case _: OffsetOutOfRangeException => // This is good.
    }

    assertEquals("Reading from below the specified maxOffset should produce 0 byte read.", 0, log.read(1025, 1000, Some(1024)).messageSet.sizeInBytes)
  }

  /**
   * Test that covers reads and writes on a multisegment log. This test appends a bunch of messages
   * and then reads them all back and checks that the message read and offset matches what was appended.
   */
  @Test
  def testLogRolls() {
    /* create a multipart log with 100 messages */
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 100: java.lang.Integer)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
    val numMessages = 100
    val messageSets = (0 until numMessages).map(i => TestUtils.singleMessageSet(i.toString.getBytes))
    messageSets.foreach(log.append(_))
    log.flush

    /* do successive reads to ensure all our messages are there */
    var offset = 0L
    for(i <- 0 until numMessages) {
      val messages = log.read(offset, 1024*1024).messageSet
      assertEquals("Offsets not equal", offset, messages.head.offset)
      assertEquals("Messages not equal at offset " + offset, messageSets(i).head.message,
        messages.head.message.toFormatVersion(messageSets(i).head.message.magic))
      offset = messages.head.offset + 1
    }
    val lastRead = log.read(startOffset = numMessages, maxLength = 1024*1024, maxOffset = Some(numMessages + 1)).messageSet
    assertEquals("Should be no more messages", 0, lastRead.size)

    // check that rolling the log forced a flushed the log--the flush is asyn so retry in case of failure
    TestUtils.retry(1000L){
      assertTrue("Log role should have forced flush", log.recoveryPoint >= log.activeSegment.baseOffset)
    }
  }

  /**
   * Test reads at offsets that fall within compressed message set boundaries.
   */
  @Test
  def testCompressedMessages() {
    /* this log should roll after every messageset */
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 100: java.lang.Integer)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)

    /* append 2 compressed message sets, each with two messages giving offsets 0, 1, 2, 3 */
    log.append(new ByteBufferMessageSet(DefaultCompressionCodec, new Message("hello".getBytes), new Message("there".getBytes)))
    log.append(new ByteBufferMessageSet(DefaultCompressionCodec, new Message("alpha".getBytes), new Message("beta".getBytes)))

    def read(offset: Int) = ByteBufferMessageSet.deepIterator(log.read(offset, 4096).messageSet.head)

    /* we should always get the first message in the compressed set when reading any offset in the set */
    assertEquals("Read at offset 0 should produce 0", 0, read(0).next().offset)
    assertEquals("Read at offset 1 should produce 0", 0, read(1).next().offset)
    assertEquals("Read at offset 2 should produce 2", 2, read(2).next().offset)
    assertEquals("Read at offset 3 should produce 2", 2, read(3).next().offset)
  }

  /**
   * Test garbage collecting old segments
   */
  @Test
  def testThatGarbageCollectingSegmentsDoesntChangeOffset() {
    for(messagesToAppend <- List(0, 1, 25)) {
      logDir.mkdirs()
      // first test a log segment starting at 0
      val logProps = new Properties()
      logProps.put(LogConfig.SegmentBytesProp, 100: java.lang.Integer)
      logProps.put(LogConfig.RetentionMsProp, 0: java.lang.Integer)
      val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)
      for(i <- 0 until messagesToAppend)
        log.append(TestUtils.singleMessageSet(payload = i.toString.getBytes, timestamp = time.milliseconds - 10))

      val currOffset = log.logEndOffset
      assertEquals(currOffset, messagesToAppend)

      // time goes by; the log file is deleted
      log.deleteOldSegments()

      assertEquals("Deleting segments shouldn't have changed the logEndOffset", currOffset, log.logEndOffset)
      assertEquals("We should still have one segment left", 1, log.numberOfSegments)
      assertEquals("Further collection shouldn't delete anything", 0, log.deleteOldSegments())
      assertEquals("Still no change in the logEndOffset", currOffset, log.logEndOffset)
      assertEquals("Should still be able to append and should get the logEndOffset assigned to the new append",
                   currOffset,
                   log.append(TestUtils.singleMessageSet("hello".getBytes)).firstOffset)

      // cleanup the log
      log.delete()
    }
  }

  /**
   *  MessageSet size shouldn't exceed the config.segmentSize, check that it is properly enforced by
   * appending a message set larger than the config.segmentSize setting and checking that an exception is thrown.
   */
  @Test
  def testMessageSetSizeCheck() {
    val messageSet = new ByteBufferMessageSet(NoCompressionCodec, new Message ("You".getBytes), new Message("bethe".getBytes))
    // append messages to log
    val configSegmentSize = messageSet.sizeInBytes - 1
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, configSegmentSize: java.lang.Integer)
    // We use need to use magic value 1 here because the test is message size sensitive.
    logProps.put(LogConfig.MessageFormatVersionProp, ApiVersion.latestVersion.toString)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)

    try {
      log.append(messageSet)
      fail("message set should throw RecordBatchTooLargeException.")
    } catch {
      case _: RecordBatchTooLargeException => // this is good
    }
  }

  @Test
  def testCompactedTopicConstraints() {
    val keyedMessage = new Message(bytes = "this message has a key".getBytes, key = "and here it is".getBytes, Message.NoTimestamp, Message.CurrentMagicValue)
    val anotherKeyedMessage = new Message(bytes = "this message also has a key".getBytes, key ="another key".getBytes, Message.NoTimestamp, Message.CurrentMagicValue)
    val unkeyedMessage = new Message(bytes = "this message does not have a key".getBytes)

    val messageSetWithUnkeyedMessage = new ByteBufferMessageSet(NoCompressionCodec, unkeyedMessage, keyedMessage)
    val messageSetWithOneUnkeyedMessage = new ByteBufferMessageSet(NoCompressionCodec, unkeyedMessage)
    val messageSetWithCompressedKeyedMessage = new ByteBufferMessageSet(GZIPCompressionCodec, keyedMessage)
    val messageSetWithCompressedUnkeyedMessage = new ByteBufferMessageSet(GZIPCompressionCodec, keyedMessage, unkeyedMessage)

    val messageSetWithKeyedMessage = new ByteBufferMessageSet(NoCompressionCodec, keyedMessage)
    val messageSetWithKeyedMessages = new ByteBufferMessageSet(NoCompressionCodec, keyedMessage, anotherKeyedMessage)

    val logProps = new Properties()
    logProps.put(LogConfig.CleanupPolicyProp, LogConfig.Compact)

    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time)

    try {
      log.append(messageSetWithUnkeyedMessage)
      fail("Compacted topics cannot accept a message without a key.")
    } catch {
      case _: CorruptRecordException => // this is good
    }
    try {
      log.append(messageSetWithOneUnkeyedMessage)
      fail("Compacted topics cannot accept a message without a key.")
    } catch {
      case _: CorruptRecordException => // this is good
    }
    try {
      log.append(messageSetWithCompressedUnkeyedMessage)
      fail("Compacted topics cannot accept a message without a key.")
    } catch {
      case _: CorruptRecordException => // this is good
    }

    // the following should succeed without any InvalidMessageException
    log.append(messageSetWithKeyedMessage)
    log.append(messageSetWithKeyedMessages)
    log.append(messageSetWithCompressedKeyedMessage)
  }

  /**
   * We have a max size limit on message appends, check that it is properly enforced by appending a message larger than the
   * setting and checking that an exception is thrown.
   */
  @Test
  def testMessageSizeCheck() {
    val first = new ByteBufferMessageSet(NoCompressionCodec, new Message ("You".getBytes), new Message("bethe".getBytes))
    val second = new ByteBufferMessageSet(NoCompressionCodec, new Message("change (I need more bytes)".getBytes))

    // append messages to log
    val maxMessageSize = second.sizeInBytes - 1
    val logProps = new Properties()
    logProps.put(LogConfig.MaxMessageBytesProp, maxMessageSize: java.lang.Integer)
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, time.scheduler, time = time)

    // should be able to append the small message
    log.append(first)

    try {
      log.append(second)
      fail("Second message set should throw MessageSizeTooLargeException.")
    } catch {
      case _: RecordTooLargeException => // this is good
    }
  }
  /**
   * Append a bunch of messages to a log and then re-open it both with and without recovery and check that the log re-initializes correctly.
   */
  @Test
  def testLogRecoversToCorrectOffset() {
    val numMessages = 100
    val messageSize = 100
    val segmentSize = 7 * messageSize
    val indexInterval = 3 * messageSize
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, segmentSize: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, indexInterval: java.lang.Integer)
    logProps.put(LogConfig.SegmentIndexBytesProp, 4096: java.lang.Integer)
    val config = LogConfig(logProps)
    var log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)
    for(i <- 0 until numMessages)
      log.append(TestUtils.singleMessageSet(payload = TestUtils.randomBytes(messageSize),
        timestamp = time.milliseconds + i * 10))
    assertEquals("After appending %d messages to an empty log, the log end offset should be %d".format(numMessages, numMessages), numMessages, log.logEndOffset)
    val lastIndexOffset = log.activeSegment.index.lastOffset
    val numIndexEntries = log.activeSegment.index.entries
    val lastOffset = log.logEndOffset
    // After segment is closed, the last entry in the time index should be (largest timestamp -> last offset).
    val lastTimeIndexOffset = log.logEndOffset - 1
    val lastTimeIndexTimestamp  = log.activeSegment.largestTimestamp
    // Depending on when the last time index entry is inserted, an entry may or may not be inserted into the time index.
    val numTimeIndexEntries = log.activeSegment.timeIndex.entries + {
      if (log.activeSegment.timeIndex.lastEntry.offset == log.logEndOffset - 1) 0 else 1
    }
    log.close()

    def verifyRecoveredLog(log: Log) {
      assertEquals(s"Should have $numMessages messages when log is reopened w/o recovery", numMessages, log.logEndOffset)
      assertEquals("Should have same last index offset as before.", lastIndexOffset, log.activeSegment.index.lastOffset)
      assertEquals("Should have same number of index entries as before.", numIndexEntries, log.activeSegment.index.entries)
      assertEquals("Should have same last time index timestamp", lastTimeIndexTimestamp, log.activeSegment.timeIndex.lastEntry.timestamp)
      assertEquals("Should have same last time index offset", lastTimeIndexOffset, log.activeSegment.timeIndex.lastEntry.offset)
      assertEquals("Should have same number of time index entries as before.", numTimeIndexEntries, log.activeSegment.timeIndex.entries)
    }

    log = new Log(logDir, config, recoveryPoint = lastOffset, time.scheduler, time)
    verifyRecoveredLog(log)
    log.close()

    // test recovery case
    log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)
    verifyRecoveredLog(log)
    log.close()
  }

  /**
   * Test building the time index on the follower by setting assignOffsets to false.
   */
  @Test
  def testBuildTimeIndexWhenNotAssigningOffsets() {
    val numMessages = 100
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 10000: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)

    val config = LogConfig(logProps)
    val log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)

    val messages = (0 until numMessages).map { i =>
      new ByteBufferMessageSet(NoCompressionCodec, new LongRef(100 + i), new Message(i.toString.getBytes(), time.milliseconds + i, Message.MagicValue_V1))
    }
    messages.foreach(log.append(_, assignOffsets = false))
    val timeIndexEntries = log.logSegments.foldLeft(0) { (entries, segment) => entries + segment.timeIndex.entries }
    assertEquals(s"There should be ${numMessages - 1} time index entries", numMessages - 1, timeIndexEntries)
    assertEquals(s"The last time index entry should have timestamp ${time.milliseconds + numMessages - 1}",
      time.milliseconds + numMessages - 1, log.activeSegment.timeIndex.lastEntry.timestamp)
  }

  /**
   * Test that if we manually delete an index segment it is rebuilt when the log is re-opened
   */
  @Test
  def testIndexRebuild() {
    // publish the messages and close the log
    val numMessages = 200
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 200: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)

    val config = LogConfig(logProps)
    var log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)
    for(i <- 0 until numMessages)
      log.append(TestUtils.singleMessageSet(payload = TestUtils.randomBytes(10), timestamp = time.milliseconds + i * 10))
    val indexFiles = log.logSegments.map(_.index.file)
    val timeIndexFiles = log.logSegments.map(_.timeIndex.file)
    log.close()

    // delete all the index files
    indexFiles.foreach(_.delete())
    timeIndexFiles.foreach(_.delete())

    // reopen the log
    log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)
    assertEquals("Should have %d messages when log is reopened".format(numMessages), numMessages, log.logEndOffset)
    assertTrue("The index should have been rebuilt", log.logSegments.head.index.entries > 0)
    assertTrue("The time index should have been rebuilt", log.logSegments.head.timeIndex.entries > 0)
    for(i <- 0 until numMessages) {
      assertEquals(i, log.read(i, 100, None).messageSet.head.offset)
      if (i == 0)
        assertEquals(log.logSegments.head.baseOffset, log.fetchOffsetsByTimestamp(time.milliseconds + i * 10).get.offset)
      else
        assertEquals(i, log.fetchOffsetsByTimestamp(time.milliseconds + i * 10).get.offset)
    }
    log.close()
  }

  /**
   * Test that if messages format version of the messages in a segment is before 0.10.0, the time index should be empty.
   */
  @Test
  def testRebuildTimeIndexForOldMessages() {
    val numMessages = 200
    val segmentSize = 200
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, segmentSize: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)
    logProps.put(LogConfig.MessageFormatVersionProp, "0.9.0")

    val config = LogConfig(logProps)
    var log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)
    for(i <- 0 until numMessages)
      log.append(TestUtils.singleMessageSet(payload = TestUtils.randomBytes(10), timestamp = time.milliseconds + i * 10))
    val timeIndexFiles = log.logSegments.map(_.timeIndex.file)
    log.close()

    // Delete the time index.
    timeIndexFiles.foreach(_.delete())

    // The rebuilt time index should be empty
    log = new Log(logDir, config, recoveryPoint = numMessages + 1, time.scheduler, time)
    val segArray = log.logSegments.toArray
    for (i <- 0 until segArray.size - 1)
      assertEquals("The time index should be empty", 0, segArray(i).timeIndex.entries)

  }

  /**
   * Test that if we have corrupted an index segment it is rebuilt when the log is re-opened
   */
  @Test
  def testCorruptIndexRebuild() {
    // publish the messages and close the log
    val numMessages = 200
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 200: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)

    val config = LogConfig(logProps)
    var log = new Log(logDir, config, recoveryPoint = 0L, time.scheduler, time)
    for(i <- 0 until numMessages)
      log.append(TestUtils.singleMessageSet(payload = TestUtils.randomBytes(10), timestamp = time.milliseconds + i * 10))
    val indexFiles = log.logSegments.map(_.index.file)
    val timeIndexFiles = log.logSegments.map(_.timeIndex.file)
    log.close()

    // corrupt all the index files
    for( file <- indexFiles) {
      val bw = new BufferedWriter(new FileWriter(file))
      bw.write("  ")
      bw.close()
    }

    // corrupt all the index files
    for( file <- timeIndexFiles) {
      val bw = new BufferedWriter(new FileWriter(file))
      bw.write("  ")
      bw.close()
    }

    // reopen the log
    log = new Log(logDir, config, recoveryPoint = 200L, time.scheduler, time)
    assertEquals("Should have %d messages when log is reopened".format(numMessages), numMessages, log.logEndOffset)
    for(i <- 0 until numMessages) {
      assertEquals(i, log.read(i, 100, None).messageSet.head.offset)
      if (i == 0)
        assertEquals(log.logSegments.head.baseOffset, log.fetchOffsetsByTimestamp(time.milliseconds + i * 10).get.offset)
      else
        assertEquals(i, log.fetchOffsetsByTimestamp(time.milliseconds + i * 10).get.offset)
    }
    log.close()
  }

  /**
   * Test the Log truncate operations
   */
  @Test
  def testTruncateTo() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val setSize = set.sizeInBytes
    val msgPerSeg = 10
    val segmentSize = msgPerSeg * setSize  // each segment will be 10 messages

    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, segmentSize: java.lang.Integer)

    // create a log
    val log = new Log(logDir, LogConfig(logProps), recoveryPoint = 0L, scheduler = time.scheduler, time = time)
    assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments)

    for (_ <- 1 to msgPerSeg)
      log.append(set)

    assertEquals("There should be exactly 1 segments.", 1, log.numberOfSegments)
    assertEquals("Log end offset should be equal to number of messages", msgPerSeg, log.logEndOffset)

    val lastOffset = log.logEndOffset
    val size = log.size
    log.truncateTo(log.logEndOffset) // keep the entire log
    assertEquals("Should not change offset", lastOffset, log.logEndOffset)
    assertEquals("Should not change log size", size, log.size)
    log.truncateTo(log.logEndOffset + 1) // try to truncate beyond lastOffset
    assertEquals("Should not change offset but should log error", lastOffset, log.logEndOffset)
    assertEquals("Should not change log size", size, log.size)
    log.truncateTo(msgPerSeg/2) // truncate somewhere in between
    assertEquals("Should change offset", log.logEndOffset, msgPerSeg/2)
    assertTrue("Should change log size", log.size < size)
    log.truncateTo(0) // truncate the entire log
    assertEquals("Should change offset", 0, log.logEndOffset)
    assertEquals("Should change log size", 0, log.size)

    for (_ <- 1 to msgPerSeg)
      log.append(set)

    assertEquals("Should be back to original offset", log.logEndOffset, lastOffset)
    assertEquals("Should be back to original size", log.size, size)
    log.truncateFullyAndStartAt(log.logEndOffset - (msgPerSeg - 1))
    assertEquals("Should change offset", log.logEndOffset, lastOffset - (msgPerSeg - 1))
    assertEquals("Should change log size", log.size, 0)

    for (_ <- 1 to msgPerSeg)
      log.append(set)

    assertTrue("Should be ahead of to original offset", log.logEndOffset > msgPerSeg)
    assertEquals("log size should be same as before", size, log.size)
    log.truncateTo(0) // truncate before first start offset in the log
    assertEquals("Should change offset", 0, log.logEndOffset)
    assertEquals("Should change log size", log.size, 0)
  }

  /**
   * Verify that when we truncate a log the index of the last segment is resized to the max index size to allow more appends
   */
  @Test
  def testIndexResizingAtTruncation() {
    val setSize = TestUtils.singleMessageSet(payload = "test".getBytes).sizeInBytes
    val msgPerSeg = 10
    val segmentSize = msgPerSeg * setSize  // each segment will be 10 messages
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, segmentSize: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, (setSize - 1): java.lang.Integer)
    val config = LogConfig(logProps)
    val log = new Log(logDir, config, recoveryPoint = 0L, scheduler = time.scheduler, time = time)
    assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments)

    for (i<- 1 to msgPerSeg)
      log.append(TestUtils.singleMessageSet(payload = "test".getBytes, timestamp = time.milliseconds + i))
    assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments)

    time.sleep(msgPerSeg)
    for (i<- 1 to msgPerSeg)
      log.append(TestUtils.singleMessageSet(payload = "test".getBytes, timestamp = time.milliseconds + i))
    assertEquals("There should be exactly 2 segment.", 2, log.numberOfSegments)
    val expectedEntries = msgPerSeg - 1

    assertEquals(s"The index of the first segment should have $expectedEntries entries", expectedEntries, log.logSegments.toList.head.index.maxEntries)
    assertEquals(s"The time index of the first segment should have $expectedEntries entries", expectedEntries, log.logSegments.toList.head.timeIndex.maxEntries)

    log.truncateTo(0)
    assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments)
    assertEquals("The index of segment 1 should be resized to maxIndexSize", log.config.maxIndexSize/8, log.logSegments.toList.head.index.maxEntries)
    assertEquals("The time index of segment 1 should be resized to maxIndexSize", log.config.maxIndexSize/12, log.logSegments.toList.head.timeIndex.maxEntries)

    time.sleep(msgPerSeg)
    for (i<- 1 to msgPerSeg)
      log.append(TestUtils.singleMessageSet(payload = "test".getBytes, timestamp = time.milliseconds + i))
    assertEquals("There should be exactly 1 segment.", 1, log.numberOfSegments)
  }

  /**
   * When we open a log any index segments without an associated log segment should be deleted.
   */
  @Test
  def testBogusIndexSegmentsAreRemoved() {
    val bogusIndex1 = Log.indexFilename(logDir, 0)
    val bogusTimeIndex1 = Log.timeIndexFilename(logDir, 0)
    val bogusIndex2 = Log.indexFilename(logDir, 5)
    val bogusTimeIndex2 = Log.timeIndexFilename(logDir, 5)

    val set = TestUtils.singleMessageSet("test".getBytes)
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer)
    logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)
    val log = new Log(logDir,
                      LogConfig(logProps),
                      recoveryPoint = 0L,
                      time.scheduler,
                      time)

    assertTrue("The first index file should have been replaced with a larger file", bogusIndex1.length > 0)
    assertTrue("The first time index file should have been replaced with a larger file", bogusTimeIndex1.length > 0)
    assertFalse("The second index file should have been deleted.", bogusIndex2.exists)
    assertFalse("The second time index file should have been deleted.", bogusTimeIndex2.exists)

    // check that we can append to the log
    for (_ <- 0 until 10)
      log.append(set)

    log.delete()
  }

  /**
   * Verify that truncation works correctly after re-opening the log
   */
  @Test
  def testReopenThenTruncate() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer)
    logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 10000: java.lang.Integer)
    val config = LogConfig(logProps)

    // create a log
    var log = new Log(logDir,
                      config,
                      recoveryPoint = 0L,
                      time.scheduler,
                      time)

    // add enough messages to roll over several segments then close and re-open and attempt to truncate
    for (_ <- 0 until 100)
      log.append(set)
    log.close()
    log = new Log(logDir,
                  config,
                  recoveryPoint = 0L,
                  time.scheduler,
                  time)
    log.truncateTo(3)
    assertEquals("All but one segment should be deleted.", 1, log.numberOfSegments)
    assertEquals("Log end offset should be 3.", 3, log.logEndOffset)
  }

  /**
   * Test that deleted files are deleted after the appropriate time.
   */
  @Test
  def testAsyncDelete() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val asyncDeleteMs = 1000
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer)
    logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 10000: java.lang.Integer)
    logProps.put(LogConfig.FileDeleteDelayMsProp, asyncDeleteMs: java.lang.Integer)
    logProps.put(LogConfig.RetentionMsProp, 0: java.lang.Integer)
    val config = LogConfig(logProps)

    val log = new Log(logDir,
                      config,
                      recoveryPoint = 0L,
                      time.scheduler,
                      time)

    // append some messages to create some segments
    for (_ <- 0 until 100)
      log.append(set)

    // files should be renamed
    val segments = log.logSegments.toArray
    val oldFiles = segments.map(_.log.file) ++ segments.map(_.index.file)
    // expire all segments
    log.logSegments.foreach(_.lastModified = time.milliseconds - 1000L)

    log.deleteOldSegments()

    assertEquals("Only one segment should remain.", 1, log.numberOfSegments)
    assertTrue("All log and index files should end in .deleted", segments.forall(_.log.file.getName.endsWith(Log.DeletedFileSuffix)) &&
                                                                 segments.forall(_.index.file.getName.endsWith(Log.DeletedFileSuffix)))
    assertTrue("The .deleted files should still be there.", segments.forall(_.log.file.exists) &&
                                                            segments.forall(_.index.file.exists))
    assertTrue("The original file should be gone.", oldFiles.forall(!_.exists))

    // when enough time passes the files should be deleted
    val deletedFiles = segments.map(_.log.file) ++ segments.map(_.index.file)
    time.sleep(asyncDeleteMs + 1)
    assertTrue("Files should all be gone.", deletedFiles.forall(!_.exists))
  }

  /**
   * Any files ending in .deleted should be removed when the log is re-opened.
   */
  @Test
  def testOpenDeletesObsoleteFiles() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer)
    logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.RetentionMsProp, 0: java.lang.Integer)
    val config = LogConfig(logProps)
    var log = new Log(logDir,
                      config,
                      recoveryPoint = 0L,
                      time.scheduler,
                      time)

    // append some messages to create some segments
    for (_ <- 0 until 100)
      log.append(set)

    // expire all segments
    log.logSegments.foreach(_.lastModified = time.milliseconds - 1000)
    log.deleteOldSegments()
    log.close()

    log = new Log(logDir,
                  config,
                  recoveryPoint = 0L,
                  time.scheduler,
                  time)
    assertEquals("The deleted segments should be gone.", 1, log.numberOfSegments)
  }

  @Test
  def testAppendMessageWithNullPayload() {
    val log = new Log(logDir,
                      LogConfig(),
                      recoveryPoint = 0L,
                      time.scheduler,
                      time)
    log.append(new ByteBufferMessageSet(new Message(bytes = null)))
    val messageSet = log.read(0, 4096, None).messageSet
    assertEquals(0, messageSet.head.offset)
    assertTrue("Message payload should be null.", messageSet.head.message.isNull)
  }

  @Test(expected = classOf[IllegalArgumentException])
  def testAppendWithOutOfOrderOffsetsThrowsException() {
    val log = new Log(logDir,
      LogConfig(),
      recoveryPoint = 0L,
      time.scheduler,
      time)
    val messages = (0 until 2).map(id => new Message(id.toString.getBytes)).toArray
    messages.foreach(message => log.append(new ByteBufferMessageSet(message)))
    val invalidMessage = new ByteBufferMessageSet(new Message(1.toString.getBytes))
    log.append(invalidMessage, assignOffsets = false)
  }

  @Test
  def testCorruptLog() {
    // append some messages to create some segments
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)
    logProps.put(LogConfig.MaxMessageBytesProp, 64*1024: java.lang.Integer)
    val config = LogConfig(logProps)
    val set = TestUtils.singleMessageSet("test".getBytes)
    val recoveryPoint = 50L
    for (_ <- 0 until 50) {
      // create a log and write some messages to it
      logDir.mkdirs()
      var log = new Log(logDir,
                        config,
                        recoveryPoint = 0L,
                        time.scheduler,
                        time)
      val numMessages = 50 + TestUtils.random.nextInt(50)
      for (_ <- 0 until numMessages)
        log.append(set)
      val messages = log.logSegments.flatMap(_.log.iterator.toList)
      log.close()

      // corrupt index and log by appending random bytes
      TestUtils.appendNonsenseToFile(log.activeSegment.index.file, TestUtils.random.nextInt(1024) + 1)
      TestUtils.appendNonsenseToFile(log.activeSegment.log.file, TestUtils.random.nextInt(1024) + 1)

      // attempt recovery
      log = new Log(logDir, config, recoveryPoint, time.scheduler, time)
      assertEquals(numMessages, log.logEndOffset)
      assertEquals("Messages in the log after recovery should be the same.", messages, log.logSegments.flatMap(_.log.iterator.toList))
      Utils.delete(logDir)
    }
  }

  @Test
  def testCleanShutdownFile() {
    // append some messages to create some segments
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.MaxMessageBytesProp, 64*1024: java.lang.Integer)
    logProps.put(LogConfig.IndexIntervalBytesProp, 1: java.lang.Integer)
    val config = LogConfig(logProps)
    val set = TestUtils.singleMessageSet("test".getBytes)
    val parentLogDir = logDir.getParentFile
    assertTrue("Data directory %s must exist", parentLogDir.isDirectory)
    val cleanShutdownFile = new File(parentLogDir, Log.CleanShutdownFile)
    cleanShutdownFile.createNewFile()
    assertTrue(".kafka_cleanshutdown must exist", cleanShutdownFile.exists())
    var recoveryPoint = 0L
    // create a log and write some messages to it
    var log = new Log(logDir,
      config,
      recoveryPoint = 0L,
      time.scheduler,
      time)
    for (_ <- 0 until 100)
      log.append(set)
    log.close()

    // check if recovery was attempted. Even if the recovery point is 0L, recovery should not be attempted as the
    // clean shutdown file exists.
    recoveryPoint = log.logEndOffset
    log = new Log(logDir, config, 0L, time.scheduler, time)
    assertEquals(recoveryPoint, log.logEndOffset)
    cleanShutdownFile.delete()
  }

  @Test
  def testParseTopicPartitionName() {
    val topic = "test_topic"
    val partition = "143"
    val dir = new File(logDir + topicPartitionName(topic, partition))
    val topicAndPartition = Log.parseTopicPartitionName(dir)
    assertEquals(topic, topicAndPartition.asTuple._1)
    assertEquals(partition.toInt, topicAndPartition.asTuple._2)
  }

  @Test
  def testParseTopicPartitionNameForEmptyName() {
    try {
      val dir = new File("")
      Log.parseTopicPartitionName(dir)
      fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
    } catch {
      case _: Exception => // its GOOD!
    }
  }

  @Test
  def testParseTopicPartitionNameForNull() {
    try {
      val dir: File = null
      Log.parseTopicPartitionName(dir)
      fail("KafkaException should have been thrown for dir: " + dir)
    } catch {
      case _: Exception => // its GOOD!
    }
  }

  @Test
  def testParseTopicPartitionNameForMissingSeparator() {
    val topic = "test_topic"
    val partition = "1999"
    val dir = new File(logDir + File.separator + topic + partition)
    try {
      Log.parseTopicPartitionName(dir)
      fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
    } catch {
      case _: Exception => // its GOOD!
    }
  }

  @Test
  def testParseTopicPartitionNameForMissingTopic() {
    val topic = ""
    val partition = "1999"
    val dir = new File(logDir + topicPartitionName(topic, partition))
    try {
      Log.parseTopicPartitionName(dir)
      fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
    } catch {
      case _: Exception => // its GOOD!
    }
  }

  @Test
  def testParseTopicPartitionNameForMissingPartition() {
    val topic = "test_topic"
    val partition = ""
    val dir = new File(logDir + topicPartitionName(topic, partition))
    try {
      Log.parseTopicPartitionName(dir)
      fail("KafkaException should have been thrown for dir: " + dir.getCanonicalPath)
    } catch {
      case _: Exception => // its GOOD!
    }
  }

  def topicPartitionName(topic: String, partition: String): String =
    File.separator + topic + "-" + partition

  @Test
  def testDeleteOldSegmentsMethod() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, set.sizeInBytes * 5: java.lang.Integer)
    logProps.put(LogConfig.SegmentIndexBytesProp, 1000: java.lang.Integer)
    logProps.put(LogConfig.RetentionMsProp, 0: java.lang.Integer)
    val config = LogConfig(logProps)
    val log = new Log(logDir,
      config,
      recoveryPoint = 0L,
      time.scheduler,
      time)

    // append some messages to create some segments
    for (_ <- 0 until 100)
      log.append(set)

    // expire all segments
    log.logSegments.foreach(_.lastModified = time.milliseconds - 1000)
    log.deleteOldSegments()
    assertEquals("The deleted segments should be gone.", 1, log.numberOfSegments)

    // append some messages to create some segments
    for (_ <- 0 until 100)
      log.append(set)

    log.delete()
    assertEquals("The number of segments should be 0", 0, log.numberOfSegments)
    assertEquals("The number of deleted segments should be zero.", 0, log.deleteOldSegments())
  }

  @Test
  def shouldDeleteSizeBasedSegments() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val log = createLog(set.sizeInBytes, retentionBytes = set.sizeInBytes * 10)

    // append some messages to create some segments
    for (_ <- 0 until 15)
      log.append(set)

    log.deleteOldSegments
    assertEquals("should have 2 segments", 2,log.numberOfSegments)
  }

  @Test
  def shouldNotDeleteSizeBasedSegmentsWhenUnderRetentionSize() {
    val set = TestUtils.singleMessageSet("test".getBytes)
    val log = createLog(set.sizeInBytes, retentionBytes = set.sizeInBytes * 15)

    // append some messages to create some segments
    for (_ <- 0 until 15)
      log.append(set)

    log.deleteOldSegments
    assertEquals("should have 3 segments", 3,log.numberOfSegments)
  }

  @Test
  def shouldDeleteTimeBasedSegmentsReadyToBeDeleted() {
    val set = TestUtils.singleMessageSet("test".getBytes, timestamp = 10)
    val log = createLog(set.sizeInBytes, retentionMs = 10000)

    // append some messages to create some segments
    for (_ <- 0 until 15)
      log.append(set)

    log.deleteOldSegments()
    assertEquals("There should be 1 segment remaining", 1, log.numberOfSegments)
  }

  @Test
  def shouldNotDeleteTimeBasedSegmentsWhenNoneReadyToBeDeleted() {
    val set = TestUtils.singleMessageSet("test".getBytes, timestamp = time.milliseconds)
    val log = createLog(set.sizeInBytes, retentionMs = 10000000)

    // append some messages to create some segments
    for (_ <- 0 until 15)
      log.append(set)

    log.deleteOldSegments()
    assertEquals("There should be 3 segments remaining", 3, log.numberOfSegments)
  }

  @Test
  def shouldNotDeleteSegmentsWhenPolicyDoesNotIncludeDelete() {
    val set = TestUtils.singleMessageSet("test".getBytes, key = "test".getBytes(), timestamp = 10L)
    val log = createLog(set.sizeInBytes,
      retentionMs = 10000,
      cleanupPolicy = "compact")

    // append some messages to create some segments
    for (_ <- 0 until 15)
      log.append(set)

    // mark oldest segment as older the retention.ms
    log.logSegments.head.lastModified = time.milliseconds - 20000

    val segments = log.numberOfSegments
    log.deleteOldSegments()
    assertEquals("There should be 3 segments remaining", segments, log.numberOfSegments)
  }

  @Test
  def shouldDeleteSegmentsReadyToBeDeletedWhenCleanupPolicyIsCompactAndDelete() {
    val set = TestUtils.singleMessageSet("test".getBytes, key = "test".getBytes,timestamp = 10L)
    val log = createLog(set.sizeInBytes,
      retentionMs = 10000,
      cleanupPolicy = "compact,delete")

    // append some messages to create some segments
    for (_ <- 0 until 15)
      log.append(set)

    log.deleteOldSegments()
    assertEquals("There should be 1 segment remaining", 1, log.numberOfSegments)
  }

  def createLog(messageSizeInBytes: Int, retentionMs: Int = -1,
                retentionBytes: Int = -1, cleanupPolicy: String = "delete"): Log = {
    val logProps = new Properties()
    logProps.put(LogConfig.SegmentBytesProp, messageSizeInBytes * 5: Integer)
    logProps.put(LogConfig.RetentionMsProp, retentionMs: Integer)
    logProps.put(LogConfig.RetentionBytesProp, retentionBytes: Integer)
    logProps.put(LogConfig.CleanupPolicyProp, cleanupPolicy)
    val config = LogConfig(logProps)
    val log = new Log(logDir,
      config,
      recoveryPoint = 0L,
      time.scheduler,
      time)
    log
  }
}

redpajama_set_name: RedPajamaGithub